ordinary_api/server/

mod.rs

use futures_util::FutureExt;
mod ops;

use axum::http::{HeaderValue, Uri};
use axum::http::{Request, header};
use axum::response::Redirect;
use axum::{Extension, extract::ConnectInfo};
use hashbrown::HashMap;
use hyper::HeaderMap;
use hyper::body::Incoming;
use hyper_util::rt::{TokioExecutor, TokioIo};
use ordinary_app::server::OrdinaryAppServer;
#[cfg(feature = "server")]
use ordinary_config::OrdinaryApiLimits;
use ordinary_monitor::Manager;
use ordinary_storage::Storage;
use ordinary_storage::saferlmdb::{self, EnvBuilder, Environment};
use parking_lot::RwLock;
use rand_chacha::rand_core::Rng;
use rand_chacha::rand_core::SeedableRng;
use sha2::{Digest, Sha256};
use std::fs::File;
use std::io::Write;
use std::process;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
use tokio_rustls::LazyConfigAcceptor;
use tokio_rustls::StartHandshake;
use tower::Layer;

use axum::Router;
use axum::http::HeaderName;
use axum::response::Response;
use axum::routing::{get, patch, post, put};

use axum::http::StatusCode;
use tower_http::classify::ServerErrorsFailureClass;
use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer};
use uuid::Uuid;

use std::net::SocketAddr;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use parking_lot::Mutex;

use rustls_acme::is_tls_alpn_challenge;
use rustls_acme::{AcmeConfig, caches::DirCache};
use tokio::io::AsyncWriteExt;
use tokio_rustls::rustls::crypto::aws_lc_rs;
use tokio_rustls::{TlsAcceptor, rustls::ServerConfig};
use tokio_stream::StreamExt;
use tower::{Service, ServiceBuilder};
use tower_http::compression::CompressionLayer;
use tower_http::decompression::RequestDecompressionLayer;
use tower_http::trace::TraceLayer;
use tracing::Span;

use ordinary_auth::{Auth, AuthClient};

use std::fs;

use crate::{api_account_claims, api_invite_claims};
use axum::middleware::Next;
use base64::{Engine as B64Engine, engine::general_purpose::URL_SAFE_NO_PAD as b64};
use getrandom::SysRng;
use hickory_client::client::Client;
use hickory_client::proto::runtime::TokioRuntimeProvider;
use hickory_client::proto::udp::UdpClientStream;
use ordinary_config::{
    AccessTokenConfig, AuthConfig, HmacTokenAlgorithm, InviteConfig, InviteMode, MfaConfig,
    OrdinaryApiConfig, OrdinaryConfig, PasswordConfig, RedactedHashAlg, RefreshTokenConfig,
    SignedTokenAlgorithm, TotpConfig,
};
use ordinary_utils::middleware::modify_etag_for_encoding;
use ordinary_utils::{
    GMT_FORMAT, HeadersDebug, LatencyDisplay, REQUEST_ID_HEADER, WrappedRedactedHashingAlg,
    get_host, redirect_service, response_for_panic, rustls_server_config, shutdown_signal,
};
use std::error::Error;
use std::pin::pin;
use time::UtcDateTime;
use tokio::sync::watch;
use tokio_rustls::rustls::server::Acceptor;
use tower_http::catch_panic::CatchPanicLayer;
use tower_http::set_header::SetResponseHeaderLayer;
use tower_http::timeout::TimeoutLayer;
use utoipa::openapi::security::{Http, HttpAuthScheme, SecurityScheme};
use utoipa::{Modify, OpenApi};
use utoipa_swagger_ui::{Config, SwaggerUi};

pub const ROOT: &str = "root";
pub const ADMIN: &str = "admin";
pub const APPLICATION: &str = "application";
pub const AUTHENTICATION: &str = "authentication";

#[derive(OpenApi)]
#[openapi(
    info(
        title = "Ordinary API Server",
        description = "
## ordinary-api

[![docs.rs](https://img.shields.io/docsrs/ordinary-api/0.6.0-pre.1)](https://docs.rs/ordinary-api/0.6.0-pre.1)
[![dependency status](https://img.shields.io/deps-rs/ordinary-api/0.6.0-pre.1)](https://deps.rs/crate/ordinary-api/0.6.0-pre.1)

### Access Token

Use the [Ordinary CLI](https://crates.io/crates/ordinary) to get an access token.

```sh
ordinary accounts access get --min 15
```
",
        contact(),
    ),
    modifiers(&Security),
    nest(
        (path = "/v1", api = OrdinaryApiOpenApi),
    ),
    tags(
        (name = APPLICATION, description = "Application API"),
        (name = ADMIN, description = "Admin API"),
        (name = ROOT, description = "Root API"),
        (name = AUTHENTICATION, description = "Authentication API"),
    )
)]
struct ApiDoc;

struct Security;

impl Modify for Security {
    fn modify(&self, openapi: &mut utoipa::openapi::OpenApi) {
        if let Some(components) = openapi.components.as_mut() {
            components.add_security_scheme(
                "access",
                SecurityScheme::Http(Http::new(HttpAuthScheme::Bearer)),
            );
        }
    }
}

#[derive(OpenApi)]
#[openapi(paths(
    ops::accounts::invite,
    ops::accounts::list,
    ops::accounts::delete,
    ops::auth::registration_start,
    ops::auth::registration_finish,
    ops::auth::login_start,
    ops::auth::login_finish,
    ops::logs::root,
    ops::logs::app,
    ops::app::deploy,
    ops::app::patch,
    ops::app::kill,
    ops::templates::upload,
    ops::content::update,
    ops::assets::write,
    ops::models::items_list,
    ops::actions::install,
))]
pub struct OrdinaryApiOpenApi;

/// "admin" 0 | "read" 1 | "write" 2 | "update" 3 | "upload" 4 | "install" 5 | "deploy" 6 | "bridge" 7 | "kill" 8 | "erase" 9
pub(crate) fn check_ordinary_auth(
    state: &OrdinaryApiServerState,
    headers: &HeaderMap,
    permission: u8,
    domain: &str,
) -> Result<String, StatusCode> {
    if let Some(auth_header) = headers.get("authorization")
        && let Ok(str_val) = auth_header.to_str()
        && let Some(b64_token) = str_val.strip_prefix("Bearer ")
        && let Ok(token) = b64.decode(b64_token)
    {
        match state.auth.verify_access_token(&token) {
            Ok((account, claims)) => {
                let span = tracing::info_span!("permission", account);

                span.in_scope(|| {
                    if account == "root" {
                        return Ok(account.to_string());
                    }

                    if domain == "root" {
                        tracing::error!("root-only operation disallowed");
                        return Err(StatusCode::UNAUTHORIZED);
                    }

                    let check_domain = claims.idx(1).as_str();

                    if check_domain != domain {
                        tracing::error!(domain, "token domain '{check_domain}' does not match");
                        return Err(StatusCode::UNAUTHORIZED);
                    }

                    let permissions = claims.idx(2).as_vector();

                    for claim_permission in &permissions {
                        if claim_permission.as_u8() == 0 || claim_permission.as_u8() == permission {
                            tracing::info!(
                                role = match claim_permission.as_u8() {
                                    0 => "admin",
                                    1 => "read",
                                    2 => "write",
                                    3 => "update",
                                    4 => "upload",
                                    5 => "install",
                                    6 => "deploy",
                                    7 => "bridge",
                                    8 => "kill",
                                    9 => "erase",
                                    _ => "invalid",
                                },
                                "granted"
                            );

                            return Ok(account.to_string());
                        }
                    }

                    tracing::error!(
                        permission = match permission {
                            0 => "admin",
                            1 => "read",
                            2 => "write",
                            3 => "update",
                            4 => "upload",
                            5 => "install",
                            6 => "deploy",
                            7 => "bridge",
                            8 => "kill",
                            9 => "erase",
                            _ => "invalid",
                        },
                        "not in token permissions"
                    );
                    Err(StatusCode::UNAUTHORIZED)
                })
            }
            Err(err) => {
                tracing::error!(%err, "access denied");
                Err(StatusCode::UNAUTHORIZED)
            }
        }
    } else {
        tracing::error!("invalid token");
        Err(StatusCode::UNAUTHORIZED)
    }
}

pub struct WrappedOrdinaryAppServer {
    port: u16,
    app: Arc<OrdinaryAppServer>,
    kill_tx: tokio::sync::broadcast::Sender<()>,
    stream_tx: tokio::sync::mpsc::UnboundedSender<(StartHandshake<TcpStream>, SocketAddr, Span)>,
}

type Dbs = HashMap<String, (Arc<Environment>, Arc<Auth>, Arc<Storage>)>;

#[allow(clippy::struct_excessive_bools)]
pub struct OrdinaryApiServerState {
    pub domain: String,
    pub secure: bool,
    pub secure_cookies: bool,
    pub log_headers: bool,
    pub log_ips: bool,
    pub log_size: bool,
    pub auth: Arc<Auth>,
    pub apps: Arc<RwLock<HashMap<String, Arc<WrappedOrdinaryAppServer>>>>,
    pub apps_dir: PathBuf,
    pub dbs: Arc<Mutex<Dbs>>,
    pub env_name: String,
    pub provision_mode: Option<ProvisionMode>,
    pub dedicated_ports: bool,
    pub server_span: Span,
    pub log_manager: Arc<Option<Manager>>,
    pub log_index: bool,
    pub limits: String,
    pub config: OrdinaryApiConfig,

    pub signal_tx: Arc<RwLock<Option<watch::Sender<()>>>>,
    pub close_rx: Arc<RwLock<Option<watch::Receiver<()>>>>,

    pub dns_client: Arc<tokio::sync::Mutex<Client>>,
}

#[derive(PartialEq, Clone)]
pub enum ProvisionMode {
    Localhost,
    Staging,
    Production,
}

pub enum SecurityMode<T: AsRef<Path>> {
    Insecure,
    Secure(T, ProvisionMode),
}

pub struct OrdinaryApiServer {
    auth: Arc<Auth>,
    _env: Arc<Environment>,
    config: OrdinaryApiConfig,
    apps: Arc<RwLock<HashMap<String, Arc<WrappedOrdinaryAppServer>>>>,
    apps_dir: PathBuf,
    log_manager: Arc<Option<Manager>>,
}

#[cfg(feature = "server")]
impl OrdinaryApiServer {
    #[allow(clippy::similar_names)]
    pub fn init(
        env_name: &str,
        domain: &str,
        password: &str,
        base_dir_path: impl AsRef<Path>,
        storage_size: usize,
        log_manager: Arc<Option<Manager>>,
    ) -> Result<(String, Vec<u8>), Box<dyn Error>> {
        let span = tracing::info_span!("init", env = %env_name, pid = process::id());

        span.in_scope(|| {
            let config = OrdinaryApiConfig {
                // todo: pass in as a flag
                public_dns_ip: [8, 8, 8, 8],
                env_name: env_name.to_string(),
                limits: OrdinaryApiLimits::default(),
            };
            let config_file = serde_json::to_string_pretty(&config)?;

            let environment_dir_path = base_dir_path.as_ref().join("environments").join(env_name);
            let data_path = environment_dir_path.join("data");

            if data_path.exists() {
                return Err("environment already initialized".into());
            }

            fs::write(environment_dir_path.join("ordinaryd.json"), config_file)?;

            let api_server =
                OrdinaryApiServer::new(env_name, domain, base_dir_path, storage_size, log_manager)?;

            let mut input = domain.as_bytes().to_vec();
            input.extend_from_slice(b"root");

            let mut password_input = input.clone();
            password_input.extend_from_slice(password.as_bytes());

            let mut hasher = Sha256::new();
            hasher.update(&password_input);
            let password = hasher.finalize().to_vec();

            let invite_token = api_server.auth.api_invite_get(domain, "root", None)?;

            let (state, reg_start_req) =
                AuthClient::registration_start_req(b"root", &password, Some(invite_token))?;
            let reg_start_res = api_server
                .auth
                .registration_start(reg_start_req, None, None)?;

            let (private_key, reg_finish_req) =
                AuthClient::registration_finish_req(b"root", &password, &state, &reg_start_res)?;
            let (reg_finish_res, ..) = api_server.auth.registration_finish(reg_finish_req, None)?;

            let (totp, _recovery_codes) = AuthClient::decrypt_totp_mfa(
                &reg_finish_res,
                private_key,
                env_name.to_string(),
                "root".into(),
            )?;

            Ok((totp.get_url(), totp.secret))
        })
    }

    pub fn new(
        env_name: &str,
        domain: &str,
        base_dir_path: impl AsRef<Path>,
        storage_size: usize,
        log_manager: Arc<Option<Manager>>,
    ) -> Result<OrdinaryApiServer, Box<dyn Error>> {
        use ordinary_config::{HmacTokenConfig, SignedTokenConfig};

        let span = tracing::info_span!("setup", env = %env_name, pid = process::id());

        span.in_scope(|| {
            let environment_dir_path = base_dir_path.as_ref().join("environments").join(env_name);

            let config: OrdinaryApiConfig = serde_json::from_str(&fs::read_to_string(
                environment_dir_path.join("ordinaryd.json"),
            )?)?;

            let data_path = environment_dir_path.join("data");

            fs::create_dir_all(&data_path)?;

            let ps = page_size::get();

            // round up to full OS page
            let remainder = storage_size % ps;
            let mapsize = (storage_size - remainder) + ps;

            tracing::info!(mapsize = %bytesize::ByteSize(mapsize as u64).display().si_short());

            let env = Arc::new(unsafe {
                let mut env_builder = EnvBuilder::new()?;
                env_builder.set_maxreaders(126)?;
                env_builder.set_mapsize(mapsize)?;
                env_builder.set_maxdbs(13)?;
                env_builder.open(
                    match data_path.to_str() {
                        Some(v) => v,
                        None => return Err("data_path not a str".into()),
                    },
                    &saferlmdb::open::Flags::empty(),
                    0o600,
                )?
            });

            let keys_dir = environment_dir_path.join("keys");
            fs::create_dir_all(&keys_dir)?;

            let auth_key_path = keys_dir.join("auth");

            let auth_key: [u8; 32] = if auth_key_path.exists() && auth_key_path.is_file() {
                let auth_key = fs::read(&auth_key_path)?;
                let auth_key: [u8; 32] = auth_key[..].try_into()?;
                auth_key
            } else {
                let mut auth_key = [0u8; 32];
                let mut rng = rand_chacha::ChaCha20Rng::try_from_rng(&mut SysRng)?;

                rng.fill_bytes(&mut auth_key[..]);

                let mut auth_key_file = File::create(auth_key_path)?;
                auth_key_file.write_all(&auth_key)?;

                auth_key
            };

            let auth = Arc::new(Auth::new(
                domain.into(),
                Some(AuthConfig {
                    password: PasswordConfig {
                        protocol: ordinary_config::PasswordProtocol::Opaque,
                    },
                    mfa: MfaConfig {
                        totp: TotpConfig {
                            template: None,
                            algorithm: ordinary_config::TotpAlgorithm::Sha1,
                        },
                    },
                    hmac_token: HmacTokenConfig {
                        algorithm: HmacTokenAlgorithm::HmacBlake2b256,
                        rotation: 60 * 60 * 24 * 7,
                    },
                    signed_token: SignedTokenConfig {
                        algorithm: SignedTokenAlgorithm::DsaEd25519,
                        rotation: 60 * 60 * 24 * 30,
                    },
                    refresh_token: RefreshTokenConfig {
                        lifetime: 60 * 60 * 24 * 7,
                    },
                    access_token: AccessTokenConfig {
                        lifetime: 60 * 60 * 24,
                        claims: api_account_claims(),
                    },
                    client_hash: ordinary_config::ClientPasswordHash::Sha256,
                    cookies_enabled: true,
                    invite: Some(InviteConfig {
                        mode: InviteMode::Viral,
                        lifetime: 60 * 60 * 24,
                        clean_interval: (30, 90),
                        claims: Some(api_invite_claims()),
                    }),
                }),
                auth_key,
                env.clone(),
            )?);

            let apps_dir = environment_dir_path.join("apps");
            fs::create_dir_all(&apps_dir)?;

            Ok(OrdinaryApiServer {
                config,
                auth,
                _env: env,
                apps: Arc::new(RwLock::new(HashMap::new())),
                apps_dir,
                log_manager,
            })
        })
    }

    #[allow(
        clippy::too_many_arguments,
        clippy::fn_params_excessive_bools,
        clippy::too_many_lines
    )]
    pub async fn start<P, F>(
        &self,
        server_span: Span,
        mode: SecurityMode<P>,
        listener: TcpListener,
        secure_cookies: bool,
        log_headers: bool,
        log_ips: bool,
        log_size: bool,
        api_domain: String,
        api_contacts: &[String],
        redirect_listener: Option<TcpListener>,
        dedicated_ports: bool,
        log_index: bool,
        redacted_hash: Option<RedactedHashAlg>,
        swagger: bool,
        signal: fn() -> F,
    ) -> Result<(), Box<dyn Error>>
    where
        P: AsRef<Path>,
        F: Future<Output = ()> + Send + 'static,
    {
        use axum::extract::State;

        let start_span = tracing::info_span!(
            "start",
            env = &self.config.env_name,
            pid = process::id(),
            domain = api_domain
        );

        let address = SocketAddr::from((self.config.public_dns_ip, 53));
        let conn = UdpClientStream::builder(address, TokioRuntimeProvider::default()).build();

        let (client, bg) = Client::connect(conn).await?;
        tokio::spawn(bg);

        let server_span_clone = server_span.clone();
        let server_span_clone2 = server_span.clone();
        let server_span_clone3 = server_span.clone();

        let api_domain_clone = api_domain.clone();

        let limits = serde_json::to_string(&self.config.limits)?;

        let state = start_span.in_scope(|| {
            Arc::new(OrdinaryApiServerState {
                domain: api_domain.clone(),
                secure: !matches!(mode, SecurityMode::Insecure),
                secure_cookies,
                log_headers,
                log_ips,
                log_size,
                auth: self.auth.clone(),
                apps: self.apps.clone(),
                apps_dir: self.apps_dir.clone(),
                dbs: Arc::new(Mutex::new(HashMap::new())),
                env_name: self.config.env_name.clone(),
                provision_mode: match &mode {
                    SecurityMode::Insecure => None,
                    SecurityMode::Secure(_, provision) => Some(provision.clone()),
                },
                dedicated_ports,
                server_span: server_span.clone(),
                log_manager: self.log_manager.clone(),
                log_index,
                limits,
                config: self.config.clone(),

                signal_tx: Arc::new(RwLock::new(None)),
                close_rx: Arc::new(RwLock::new(None)),

                dns_client: Arc::new(tokio::sync::Mutex::new(client)),
            })
        });

        let request_id = HeaderName::from_static(REQUEST_ID_HEADER);

        let redacted_hash = if let Some(redacted_hash) = redacted_hash {
            Arc::new(Some(WrappedRedactedHashingAlg(redacted_hash)))
        } else {
            Arc::new(None)
        };

        let redacted_hash_clone = redacted_hash.clone();
        let redacted_hash_clone1 = redacted_hash.clone();

        let last_modified = UtcDateTime::now();

        let last_modified_string = last_modified.format(&GMT_FORMAT)?;
        let last_modified_header = HeaderValue::from_str(last_modified_string.as_str())?;

        let ordinaryd_resource_layers = ServiceBuilder::new()
            .layer(axum::middleware::from_fn(
                move |req_headers: HeaderMap, request: axum::extract::Request, next: Next| {
                    ordinary_utils::middleware::http_cache_middleware(
                        last_modified,
                        req_headers,
                        request,
                        next,
                    )
                },
            ))
            .layer(SetResponseHeaderLayer::if_not_present(
                header::LAST_MODIFIED,
                last_modified_header.clone(),
            ))
            .layer(SetResponseHeaderLayer::if_not_present(
                header::EXPIRES,
                |_: &Response<_>| {
                    let future = UtcDateTime::now() + time::Duration::minutes(10);

                    if let Ok(formatted) = future.format(&GMT_FORMAT)
                        && let Ok(expires) = HeaderValue::from_str(formatted.as_str())
                    {
                        return Some(expires);
                    }

                    None
                },
            ))
            .layer(SetResponseHeaderLayer::if_not_present(
                header::VARY,
                HeaderValue::from_static(header::ACCEPT_ENCODING.as_str()),
            ));

        let mut service = Router::new();

        if swagger {
            // todo: rapidoc
            // todo: scalar
            // todo: redoc

            service = service
                .merge(
                    SwaggerUi::new("/.ordinary/swagger").config(Config::from("/.ordinary/openapi")),
                )
                .route_layer(ordinaryd_resource_layers.clone().layer(
                    SetResponseHeaderLayer::if_not_present(
                        header::CONTENT_SECURITY_POLICY,
                        HeaderValue::from_static(
                            "default-src 'self'; style-src 'self' 'unsafe-inline'; img-src https://img.shields.io",
                        ),
                    ),
                ));
        }

        // todo: || openapi
        if swagger {
            let api = ApiDoc::openapi().to_json()?;

            service = service.route(
                "/.ordinary/openapi",
                get(|| async move {
                    (
                        StatusCode::OK,
                        [(header::CONTENT_TYPE, "application/json")],
                        api,
                    )
                })
                .route_layer(ordinaryd_resource_layers.clone()),
            );
        }

        service = service.route(
            "/.ordinary/limits",
            get(
                |State(state): State<Arc<OrdinaryApiServerState>>| async move {
                    (
                        StatusCode::OK,
                        [(header::CONTENT_TYPE, "application/json")],
                        state.limits.clone(),
                    )
                },
            )
            .route_layer(ordinaryd_resource_layers.clone()),
        );

        let mut service = service
            .route("/healthz", get(|| async { StatusCode::OK }))
            // auth ops
            .route(
                "/v1/accounts/registration/start",
                put(ops::auth::registration_start),
            )
            .route(
                "/v1/accounts/registration/finish",
                put(ops::auth::registration_finish),
            )
            .route("/v1/accounts/login/start", put(ops::auth::login_start))
            .route("/v1/accounts/login/finish", put(ops::auth::login_finish))
            .route("/v1/accounts/access/get", put(ops::auth::access_get))
            .route("/v1/accounts/logout", put(ops::auth::logout))
            // reset password
            .route(
                "/v1/accounts/password/reset/login/start",
                put(ops::auth::password_reset_login_start),
            )
            .route(
                "/v1/accounts/password/reset/login/finish",
                put(ops::auth::password_reset_login_finish),
            )
            .route(
                "/v1/accounts/password/reset/registration/start",
                put(ops::auth::password_reset_registration_start),
            )
            .route(
                "/v1/accounts/password/reset/registration/finish",
                put(ops::auth::password_reset_registration_finish),
            )
            // forgot password
            .route(
                "/v1/accounts/password/forgot/start",
                put(ops::auth::password_forgot_start),
            )
            .route(
                "/v1/accounts/password/forgot/finish",
                put(ops::auth::password_forgot_finish),
            )
            // reset totp mfa
            .route(
                "/v1/accounts/mfa/totp/reset/start",
                put(ops::auth::mfa_reset_totp_start),
            )
            .route(
                "/v1/accounts/mfa/totp/reset/finish",
                put(ops::auth::mfa_reset_totp_finish),
            )
            // lost totp mfa
            .route(
                "/v1/accounts/mfa/totp/lost/start",
                put(ops::auth::mfa_lost_totp_start),
            )
            .route(
                "/v1/accounts/mfa/totp/lost/finish",
                put(ops::auth::mfa_lost_totp_finish),
            )
            // delete account
            .route(
                "/v1/accounts/delete/start",
                put(ops::auth::account_delete_start),
            )
            .route(
                "/v1/accounts/delete/finish",
                put(ops::auth::account_delete_finish),
            )
            // reset recovery codes
            .route(
                "/v1/accounts/recovery-codes/reset/start",
                put(ops::auth::recovery_reset_codes_start),
            )
            .route(
                "/v1/accounts/recovery-codes/reset/finish",
                put(ops::auth::recovery_reset_codes_finish),
            )
            // root ops
            .route("/v1/apps", get(ops::app::list))
            .route("/v1/logs", get(ops::logs::root))
            // admin ops
            .route("/v1/accounts/invite", put(ops::accounts::invite))
            .route("/v1/accounts", get(ops::accounts::list))
            .route("/v1/accounts/delete", put(ops::accounts::delete))
            // app account ops
            .route("/v1/app/accounts/invite", put(ops::app::invite))
            .route("/v1/app/accounts", get(ops::app::accounts_list))
            // app ops
            .route("/v1/app/deploy", post(ops::app::deploy))
            .route("/v1/app/patch", patch(ops::app::patch))
            .route("/v1/app/migrate", put(ops::app::migrate))
            .route("/v1/app/bridge", put(ops::app::bridge))
            .route("/v1/app/kill", put(ops::app::kill))
            .route("/v1/app/erase", put(ops::app::erase))
            .route("/v1/app/logs", get(ops::logs::app))
            // template ops
            .route("/v1/templates", put(ops::templates::upload))
            // content ops
            .route("/v1/content", put(ops::content::update))
            // assets ops
            .route("/v1/assets", put(ops::assets::write))
            // action ops
            .route("/v1/actions", put(ops::actions::install))
            // models ops
            .route("/v1/models/items", get(ops::models::items_list))
            .fallback(|| async { StatusCode::NOT_FOUND })
            .with_state(state.clone())
            .layer(
                ServiceBuilder::new()
                    .layer(CatchPanicLayer::custom(response_for_panic))
                    .layer(RequestDecompressionLayer::new())
                    .layer(CompressionLayer::new()),
            )
            .layer(
                ServiceBuilder::new()
                    .layer(SetRequestIdLayer::new(request_id.clone(), MakeRequestUuid))
                    .layer(
                        TraceLayer::new_for_http()
                            .make_span_with(move |req: &Request<_>| {
                                let request_id = req.headers().get(REQUEST_ID_HEADER);

                                let ip = log_ips.then(|| {
                                    req.extensions()
                                        .get::<ConnectInfo<SocketAddr>>()
                                        .map(|addr| tracing::field::display(addr.ip()))
                                });

                                let query = req.uri().query().map(tracing::field::display);

                                server_span_clone.in_scope(|| match request_id {
                                    Some(rid) => {
                                        tracing::info_span!(
                                            "admin",
                                            domain = %api_domain_clone,
                                            id = %rid
                                                .to_str()
                                                .unwrap_or(Uuid::new_v4().to_string().as_str()),
                                            ip,
                                            path = %req.uri().path(),
                                            query,
                                        )
                                    }
                                    None => {
                                        tracing::info_span!(
                                            "admin",
                                            domain = %api_domain_clone,
                                            id = %Uuid::new_v4(),
                                            ip,
                                            path = %req.uri().path(),
                                            query,
                                        )
                                    }
                                })
                            })
                            .on_request(move |req: &Request<_>, _: &Span| {
                                let hd = log_headers
                                    .then_some(HeadersDebug(req.headers(), redacted_hash.clone()));

                                #[cfg(tracing_unstable)]
                                let headers = log_headers.then_some(tracing::field::valuable(&hd));

                                #[cfg(not(tracing_unstable))]
                                let headers = log_headers.then_some(tracing::field::debug(&hd));

                                tracing::info!(
                                    version = ?req.version(),
                                    method = %req.method(),
                                    headers,
                                    "req"
                                );
                            })
                            .on_response(move |res: &Response<_>, latency: Duration, _: &Span| {
                                let hd = log_headers.then_some(HeadersDebug(
                                    res.headers(),
                                    redacted_hash_clone.clone(),
                                ));

                                #[cfg(tracing_unstable)]
                                let headers = log_headers.then_some(tracing::field::valuable(&hd));

                                #[cfg(not(tracing_unstable))]
                                let headers = log_headers.then_some(tracing::field::debug(&hd));

                                let status = res.status().as_u16();
                                let latency = LatencyDisplay(latency.as_nanos() as f64);

                                if status >= 500 {
                                    tracing::error!(status, headers, %latency, "res");
                                } else if status >= 400 {
                                    tracing::warn!(status, headers, %latency, "res");
                                } else {
                                    tracing::info!(status, headers, %latency, "res");
                                }
                            })
                            .on_failure(
                                |error: ServerErrorsFailureClass, _: Duration, _: &Span| {
                                    tracing::error!(
                                        err = %error,
                                        "fail"
                                    );
                                },
                            ),
                    )
                    .layer(TimeoutLayer::with_status_code(
                        StatusCode::REQUEST_TIMEOUT,
                        Duration::from_secs(10),
                    ))
                    .layer(PropagateRequestIdLayer::new(request_id))
                    .layer(SetResponseHeaderLayer::if_not_present(
                        header::SERVER,
                        HeaderValue::from_static("Ordinary"),
                    ))
                    .layer(SetResponseHeaderLayer::overriding(
                        header::ETAG,
                        modify_etag_for_encoding,
                    )),
            );

        if state.secure {
            service = service.layer(SetResponseHeaderLayer::if_not_present(
                header::STRICT_TRANSPORT_SECURITY,
                HeaderValue::from_static("max-age=31536000"),
            ));
        }

        if let Some(listener) = redirect_listener {
            let api_domain = api_domain.clone();

            let request_id_header = HeaderName::from_static(REQUEST_ID_HEADER);

            let local_addr = listener.local_addr()?;
            let state_clone = state.clone();

            tokio::spawn(async move {
                use axum::extract::State;

                let span = server_span_clone3;

                span.in_scope(
                    || tracing::info!(server = %"redirect", addr = %local_addr, "listening"),
                );

                let handler = async move |uri: Uri,
                                          headers: HeaderMap,
                                          State(state): State<Arc<OrdinaryApiServerState>>|
                            -> Result<Redirect, StatusCode> {
                    // todo: handle localhost redirecting

                    if let Some(host) = get_host(&headers, &uri) {
                        if host == api_domain
                            && let Some(path_and_query) = uri.path_and_query()
                        {
                            return Ok(Redirect::permanent(&format!(
                                "https://{api_domain}{}",
                                path_and_query.as_str()
                            )));
                        }

                        let apps = state.apps.read();

                        if apps.contains_key(&host)
                            && let Some(path_and_query) = uri.path_and_query()
                        {
                            return Ok(Redirect::permanent(&format!(
                                "https://{host}{}",
                                path_and_query.as_str()
                            )));
                        }
                    }

                    Err(StatusCode::NOT_FOUND)
                };

                let service = redirect_service(
                    span,
                    redacted_hash_clone1,
                    log_ips,
                    log_headers,
                    request_id_header,
                    handler,
                    state_clone,
                );

                if let Err(err) = axum::serve(
                    listener,
                    service.into_make_service_with_connect_info::<SocketAddr>(),
                )
                .with_graceful_shutdown(shutdown_signal())
                .await
                {
                    tracing::error!(%err, "server closed");
                }
            });
        }

        let local_addr = listener.local_addr()?;

        let span = server_span_clone2;

        match mode {
            SecurityMode::Insecure => {
                span.in_scope(|| {
                    span.in_scope(|| {
                        tracing::info!(
                            server = %"primary",
                            addr = %local_addr,
                            provision = %"none",
                            "listening"
                        );
                    });
                });

                self.start_app(&span, &state)?;

                axum::serve(
                    listener,
                    service.into_make_service_with_connect_info::<SocketAddr>(),
                )
                .with_graceful_shutdown(signal())
                .await?;
            }
            SecurityMode::Secure(cert_dir_path, provision) => {
                let (signal_tx, signal_rx) = watch::channel(());
                let (close_tx, close_rx) = watch::channel(());

                {
                    let mut signal_tx_lock = state.signal_tx.write();
                    *signal_tx_lock = Some(signal_tx.clone());

                    let mut close_rx_lock = state.close_rx.write();
                    *close_rx_lock = Some(close_rx.clone());
                }

                let span_clone = span.clone();
                let signal_tx_clone = state.signal_tx.clone();

                tokio::spawn(async move {
                    signal().await;

                    span_clone.in_scope(|| {
                        tracing::warn!("graceful shutdown");
                    });

                    {
                        let mut signal_tx_lock = signal_tx_clone.write();
                        *signal_tx_lock = None;
                    }

                    drop(signal_rx);
                });

                self.start_app(&span, &state)?;

                match provision {
                    ProvisionMode::Localhost => {
                        let cert_dir_path = cert_dir_path.as_ref();

                        ordinary_utils::generate_self_signed_localhost_certs(cert_dir_path)?;

                        let rustls_config = rustls_server_config(
                            cert_dir_path.join("key.pem"),
                            cert_dir_path.join("crt.pem"),
                        )?;
                        let tls_acceptor = TlsAcceptor::from(rustls_config);

                        span.in_scope(|| {
                            tracing::info!(
                                server = %"primary",
                                addr = %local_addr,
                                provision = %"localhost",
                                "listening"
                            );
                        });

                        loop {
                            let span = span.clone();
                            let (cnx, addr) = tokio::select! {
                                conn = listener.accept() => conn?,
                                () = signal_tx.closed() => {
                                    span.in_scope(|| {
                                       tracing::warn!("not accepting new connections");
                                    });
                                    break;
                                }
                            };

                            let tower_service = service.clone();
                            let tls_acceptor = tls_acceptor.clone();

                            match tls_acceptor.accept(cnx).await {
                                Ok(stream) => {
                                    let stream = TokioIo::new(stream);

                                    let hyper_service = hyper::service::service_fn(
                                        move |request: Request<Incoming>| {
                                            let mut tower_service = Extension(ConnectInfo(addr))
                                                .layer(tower_service.clone());
                                            tower_service.call(request)
                                        },
                                    );

                                    let mut builder = hyper_util::server::conn::auto::Builder::new(
                                        TokioExecutor::new(),
                                    );
                                    builder.http2().enable_connect_protocol();

                                    let signal_tx = signal_tx.clone();
                                    let close_rx = close_rx.clone();

                                    tokio::spawn(async move {
                                        let mut conn =
                                            pin!(builder.serve_connection_with_upgrades(
                                                stream,
                                                hyper_service
                                            ));
                                        let mut signal_closed = pin!(signal_tx.closed().fuse());

                                        loop {
                                            tokio::select! {
                                                result = conn.as_mut() => {
                                                    span.in_scope(|| {
                                                        match result {
                                                            Ok(()) => {
                                                                if log_ips {
                                                                    tracing::info!(ip = %addr.ip(), port = addr.port(), "closed");
                                                                } else {
                                                                    tracing::info!("closed");
                                                                }
                                                            },
                                                            Err(err) => {
                                                                if log_ips {
                                                                    tracing::warn!(ip = %addr.ip(), port = addr.port(), %err, "connection");
                                                                } else {
                                                                    tracing::warn!(%err, "connection");
                                                                }
                                                            },
                                                        }
                                                    });

                                                    break;
                                                }
                                                () = &mut signal_closed => {
                                                    span.in_scope(|| {
                                                        tracing::warn!("shutdown");
                                                    });
                                                    conn.as_mut().graceful_shutdown();
                                                }
                                            }
                                        }

                                        drop(close_rx);
                                    });
                                }
                                Err(err) => {
                                    if log_ips {
                                        span.in_scope(|| {
                                            tracing::warn!(ip = %addr.ip(), port = addr.port(), %err, "accept");
                                        });
                                    } else {
                                        span.in_scope(|| {
                                            tracing::warn!(%err, "accept");
                                        });
                                    }
                                }
                            }
                        }
                    }
                    ProvisionMode::Staging | ProvisionMode::Production => {
                        let contacts = api_contacts
                            .iter()
                            .map(|c| format!("mailto:{c}"))
                            .collect::<Vec<String>>();

                        let cache_path = Path::new(cert_dir_path.as_ref()).join("cache");

                        let mut state = AcmeConfig::new([&api_domain])
                            .contact(contacts)
                            .cache_option(Some(DirCache::new(cache_path)))
                            .directory_lets_encrypt(provision == ProvisionMode::Production)
                            .state();

                        let challenge_rustls_config = state.challenge_rustls_config();
                        let mut default_rustls_config = ServerConfig::builder_with_provider(
                            Arc::new(aws_lc_rs::default_provider()),
                        )
                        .with_safe_default_protocol_versions()?
                        .with_no_client_auth()
                        .with_cert_resolver(state.resolver());

                        default_rustls_config.alpn_protocols =
                            vec![b"h2".to_vec(), b"http/1.1".to_vec()];

                        let default_rustls_config = Arc::new(default_rustls_config);

                        let api_domain_clone = api_domain.clone();
                        let span_clone = span.clone();

                        let provision_str = match provision {
                            ProvisionMode::Production => "production",
                            ProvisionMode::Staging => "staging",
                            _ => unreachable!(),
                        };

                        let acme_span = span_clone.in_scope(|| {
                            tracing::info_span!(
                                "acme",
                                domain = %api_domain_clone,
                                provision = %provision_str
                            )
                        });

                        let acme_span_clone = acme_span.clone();
                        tokio::spawn(async move {
                            loop {
                                match match state.next().await {
                                    Some(v) => v,
                                    None => break,
                                } {
                                    Ok(evt) => acme_span_clone.in_scope(|| tracing::info!(?evt)),
                                    Err(err) => acme_span_clone.in_scope(|| tracing::error!(?err)),
                                }
                            }
                        });

                        span.in_scope(|| {
                            tracing::info!(
                                server = %"primary",
                                addr = %local_addr,
                                provision = %provision_str,
                                "listening"
                            );
                        });

                        let api_domain = Arc::new(api_domain);

                        loop {
                            let span = span.clone();
                            let (cnx, addr) = tokio::select! {
                                conn = listener.accept() => conn?,
                                () = signal_tx.closed() => {
                                    span.in_scope(|| {
                                       tracing::warn!("not accepting new connections");
                                    });
                                    break;
                                }
                            };

                            let tower_service = service.clone();
                            let acme_span = acme_span.clone();

                            let challenge_rustls_config = challenge_rustls_config.clone();
                            let default_rustls_config = default_rustls_config.clone();

                            let api_domain = api_domain.clone();
                            let apps = self.apps.clone();

                            match LazyConfigAcceptor::new(Acceptor::default(), cnx).await {
                                Ok(start_handshake) => {
                                    if let Some(sni) = start_handshake.client_hello().server_name()
                                    {
                                        let tls_span = span.in_scope(|| {
                                            if log_ips {
                                                tracing::info_span!("tls", %sni, ip = %addr.ip(), port = addr.port())
                                            } else {
                                                tracing::info_span!("tls", %sni)
                                            }
                                        });

                                        if sni == *api_domain {
                                            if is_tls_alpn_challenge(
                                                &start_handshake.client_hello(),
                                            ) {
                                                acme_span.in_scope(|| {
                                                    tracing::info!("TLS-ALPN-01 validation");
                                                });

                                                match start_handshake
                                                    .into_stream(challenge_rustls_config)
                                                    .await
                                                {
                                                    Ok(mut tls) => match tls.shutdown().await {
                                                        Ok(()) => {}
                                                        Err(err) => tls_span.in_scope(
                                                            || tracing::error!(%err, "shutdown"),
                                                        ),
                                                    },
                                                    Err(err) => {
                                                        tls_span.in_scope(
                                                            || tracing::error!(%err, "stream"),
                                                        );
                                                    }
                                                }
                                            } else {
                                                tls_span.in_scope(|| tracing::info!("connecting"));

                                                match start_handshake
                                                    .into_stream(default_rustls_config)
                                                    .await
                                                {
                                                    Ok(as_stream) => {
                                                        let stream = TokioIo::new(as_stream);

                                                        let hyper_service = hyper::service::service_fn(
                                                            move |request: Request<Incoming>| {
                                                                let mut tower_service =
                                                                    Extension(ConnectInfo(addr))
                                                                        .layer(tower_service.clone());
                                                                tower_service.call(request)
                                                            },
                                                        );

                                                        let mut builder =
                                                            hyper_util::server::conn::auto::Builder::new(
                                                                TokioExecutor::new(),
                                                            );
                                                        builder.http2().enable_connect_protocol();

                                                        let signal_tx = signal_tx.clone();
                                                        let close_rx = close_rx.clone();
                                                        let tls_span = span.clone();

                                                        tokio::spawn(async move {
                                                            let mut conn = pin!(
                                                                builder
                                                                    .serve_connection_with_upgrades(
                                                                        stream,
                                                                        hyper_service
                                                                    )
                                                            );
                                                            let mut signal_closed =
                                                                pin!(signal_tx.closed().fuse());

                                                            loop {
                                                                tokio::select! {
                                                                    result = conn.as_mut() => {
                                                                        match result {
                                                                            Ok(()) => tls_span.in_scope(|| tracing::info!("closed")),
                                                                            Err(err) => tls_span.in_scope(|| tracing::warn!(%err, "connection")),
                                                                        }

                                                                        break;
                                                                    }
                                                                    () = &mut signal_closed => {
                                                                        tls_span.in_scope(|| tracing::warn!("shutdown"));
                                                                        conn.as_mut().graceful_shutdown();
                                                                    }
                                                                }
                                                            }

                                                            drop(close_rx);
                                                        });
                                                    }
                                                    Err(err) => {
                                                        tls_span.in_scope(
                                                            || tracing::warn!(%err, "handshake"),
                                                        );
                                                    }
                                                }
                                            }
                                        } else {
                                            let apps = apps.read();

                                            if let Some(app) = apps.get(sni) {
                                                if let Err(err) = app.stream_tx.send((
                                                    start_handshake,
                                                    addr,
                                                    tls_span.clone(),
                                                )) {
                                                    tls_span.in_scope(
                                                        || tracing::error!(%err, "handoff"),
                                                    );
                                                }
                                            } else {
                                                tls_span.in_scope(|| tracing::warn!("no match"));
                                            }
                                        }
                                    } else {
                                        let tls_span = span.in_scope(|| {
                                            if log_ips {
                                                tracing::info_span!("tls", ip = %addr.ip(), port = addr.port())
                                            } else {
                                                tracing::info_span!("tls")
                                            }
                                        });

                                        tls_span.in_scope(|| tracing::warn!("no sni"));
                                    }
                                }
                                Err(err) => {
                                    if log_ips {
                                        let tls_span = span.in_scope(|| { tracing::error_span!("tls", ip = %addr.ip(), port = addr.port()) });

                                        tls_span.in_scope(|| tracing::error!(%err, "accept"));
                                    } else {
                                        let tls_span =
                                            span.in_scope(|| tracing::error_span!("tls"));

                                        tls_span.in_scope(|| tracing::error!(%err, "accept"));
                                    }
                                }
                            }
                        }
                    }
                }

                {
                    let mut close_rx_lock = state.close_rx.write();
                    *close_rx_lock = None;
                }

                drop(close_rx);
                drop(listener);

                span.in_scope(|| {
                    tracing::warn!(
                        "waiting for {} task(s) to finish",
                        close_tx.receiver_count()
                    );
                });
                close_tx.closed().await;
            }
        }

        Ok(())
    }

    fn start_app(
        &self,
        server_span: &Span,
        state: &Arc<OrdinaryApiServerState>,
    ) -> Result<(), Box<dyn Error>> {
        let state_clone = state.clone();

        let Ok(shared_rt) = tokio::runtime::Handle::try_current() else {
            return Err("failed to get shared runtime".into());
        };

        server_span.in_scope(|| -> Result<(), Box<dyn Error>> {
            // todo: sleep this for 2 seconds or something to wait for the server to get started up

            let mut apps = state_clone.apps.write();

            for entry in fs::read_dir(&self.apps_dir)?.flatten() {
                let config_path = entry.path().join("ordinary.json");

                match fs::read_to_string(&config_path) {
                    Ok(config_str) => match serde_json::from_str::<OrdinaryConfig>(&config_str) {
                        Ok(config) => {
                            let domain = config.domain.clone();
                            let app_span = tracing::info_span!("app", %domain);

                            match ops::app::dbs(&state_clone, &config, &entry.path(), Some(app_span.clone())) {
                                Ok((auth, storage)) => {
                                    match ops::app::start(
                                        &state_clone,
                                        &config,
                                        auth,
                                        storage,
                                        match state_clone
                                            .provision_mode
                                            .as_ref()
                                            .unwrap_or(&ProvisionMode::Localhost)
                                        {
                                            ProvisionMode::Localhost => {
                                                ordinary_app::server::ProvisionMode::Localhost
                                            }
                                            ProvisionMode::Staging => {
                                                ordinary_app::server::ProvisionMode::Staging
                                            }
                                            ProvisionMode::Production => {
                                                ordinary_app::server::ProvisionMode::Production
                                            }
                                        },
                                        entry.path().join("certs"),
                                        shared_rt.clone(),
                                        Some(app_span),
                                    ) {
                                        Ok((port, app, kill_tx, stream_tx)) => {
                                            let cnames = app.config.cnames.clone();

                                            let app = Arc::new(WrappedOrdinaryAppServer {
                                                port,
                                                app,
                                                kill_tx,
                                                stream_tx,
                                            });

                                            if let Some(cnames) = cnames {
                                                for cname in cnames {
                                                    apps.insert(cname, app.clone());
                                                }
                                            }

                                            apps.insert(
                                                domain,
                                                app,
                                            );
                                        }
                                        Err(err) => {
                                            tracing::error!(%err, domain, "failed to start app");
                                        }
                                    }
                                }
                                Err(err) => {
                                    tracing::error!(%err, domain, "failed to init dbs");
                                }
                            }
                        }
                        Err(err) => {
                            if let Some(path_str) = config_path.to_str() {
                                tracing::error!(%err, path = path_str, "failed to parse config file");
                            } else {
                                tracing::error!(%err, "failed to parse config file");
                            }
                        }
                    },
                    Err(err) => {
                        if let Some(path_str) = config_path.to_str() {
                            tracing::error!(%err, path = path_str, "failed to read config file");
                        } else {
                            tracing::error!(%err, "failed to read config file");
                        }
                    }
                }
            }

            drop(apps);

            Ok(())
        })
    }
}