b436282
mod assets;
use crate::cache::Cache;
use crate::config::Config;
use crate::config::app::AppRegistry;
use crate::config::auth::AuthConfig;
use crate::config::jwt::JwtConfig;
use crate::config::pow::PowConfig;
use crate::config::redis::RedisConfig;
use crate::config::rocketmq::RocketMqConfig;
use crate::config::server::ServerConfig;
use crate::config::session::SessionConfig;
use crate::config::web::WebConfig;
use crate::db::DbConn;
use crate::email::EmailServiceBuilder;
use crate::events::EventHandler;
use crate::queue::RocketMqConn;
use crate::server::AuthSvc;
use crate::session::{RedisStore, SessionMiddleware};
use actix_web::{App, HttpServer, web};
use anyhow::{Context, Result};
use std::net::{IpAddr, SocketAddr};
const APP: &str = "oauth";
/// 邮件订阅者排空时的宽限时间。
const EMAIL_SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(10);
const GRPC_SHUTDOWN_GRACE_SECS: u64 = 10;
#[derive(Clone)]
struct AppState {
service: AuthSvc,
session: SessionConfig,
store: RedisStore,
}
/// host 网络部署下绑非回环地址即把单密钥门禁的 gRPC 面暴露公网；
/// 显式设 KENOAUTH_ALLOW_PUBLIC_BIND=1 才放行。
fn ensure_public_bind_allowed(addr: SocketAddr, label: &str) -> Result<SocketAddr> {
let loopback = match addr.ip() {
IpAddr::V4(ip) => ip.is_loopback(),
IpAddr::V6(ip) => ip.is_loopback(),
};
if loopback || std::env::var("KENOAUTH_ALLOW_PUBLIC_BIND").as_deref() == Ok("1") {
return Ok(addr);
}
Err(anyhow::anyhow!(
"{label} binds a non-loopback address ({addr}); the grpc plane is guarded by a single \
shared rpc_key. Set KENOAUTH_ALLOW_PUBLIC_BIND=1 to confirm public exposure."
))
}
pub async fn start(config: &Config, db: DbConn) -> Result<Runtime> {
let server: ServerConfig = config.section()?;
crate::jwt::service::install(config.section()?).context("loading service identity")?;
let auth_config: AuthConfig = config.section()?;
let jwt_config: JwtConfig = config.section()?;
let pow_config: PowConfig = config.section()?;
let apps: AppRegistry = config.section_or_default()?;
let redis: RedisConfig = config.section()?;
let session_config: SessionConfig = config.section()?;
let rocketmq: RocketMqConfig = config.section()?;
let web: WebConfig = config.section_or_default()?;
let email_pool: crate::config::email::EmailPoolConfig = config.section_or_default()?;
auth_config
.validate()
.map_err(anyhow::Error::msg)
.context("invalid auth configuration")?;
pow_config
.validate()
.map_err(anyhow::Error::msg)
.context("invalid pow configuration")?;
apps.validate()
.map_err(anyhow::Error::msg)
.context("invalid apps configuration")?;
redis
.validate()
.map_err(anyhow::Error::msg)
.context("invalid redis configuration")?;
rocketmq
.validate()
.map_err(anyhow::Error::msg)
.context("invalid rocketmq configuration")?;
email_pool
.is_valid()
.map_err(anyhow::Error::msg)
.context("invalid email configuration")?;
web.is_valid()
.map_err(anyhow::Error::msg)
.context("invalid web configuration")?;
crate::config::session::ensure_secure_cookie(
session_config.secure,
server.http_addr,
std::env::var("KENOAUTH_ALLOW_INSECURE_COOKIE").as_deref() == Ok("1"),
)
.map_err(anyhow::Error::msg)
.context("invalid session configuration")?;
let cors = crate::envelope::Cors::new(&web);
let cache = Cache::connect(&redis)
.await
.context("connecting to redis")?;
let mq = RocketMqConn::init(&rocketmq)
.await
.context("connecting to rocketmq")?;
let events = EventHandler::new(mq);
let mut email_builder = EmailServiceBuilder::new(email_pool)
.with_events(events.clone())
.with_database(db.clone());
if config.has("smtp") {
let smtp: crate::config::smtp::SmtpConfig = config.section()?;
email_builder = email_builder.with_transport(smtp);
}
let email_service = email_builder
.register_default_templates()
.await
.build()
.await
.context("building the email service")?;
let email_enabled = email_service.enabled();
let email_queue_depth = email_service.depth();
let email = Some(email_service);
let session_max_age_secs = session_config.max_age.max(1) as u64;
let service = AuthSvc::new(
db,
cache.clone(),
email,
Some(events),
auth_config,
pow_config,
apps,
jwt_config,
session_max_age_secs,
)
.await?;
let state = AppState {
service: service.clone(),
session: session_config,
store: RedisStore::new(cache),
};
let grpc_addr = ensure_public_bind_allowed(server.grpc_addr, "grpc")?;
let grpc_listener = tokio::net::TcpListener::bind(grpc_addr)
.await
.with_context(|| format!("binding grpc on {grpc_addr}"))?;
let grpc = crate::server::grpc::router(service.clone())
.await
.context("building the grpc router")?;
let http = HttpServer::new(move || {
App::new()
.wrap(SessionMiddleware::new(
state.session.clone(),
state.store.clone(),
))
.wrap(crate::oltp::HttpTelemetry::new("oauth"))
.wrap(crate::envelope::EnvelopeContext)
.wrap(cors.clone())
.app_data(web::Data::new(state.service.clone()))
.app_data(web::Data::new(web.clone()))
.configure(crate::server::http::router::configure)
.default_service(web::to(assets::serve))
})
.disable_signals()
.bind(ensure_public_bind_allowed(server.http_addr, "http")?)
.with_context(|| format!("binding http on {}", server.http_addr))?
.run();
let event_forwarding = service.verify_event_forwarding().await;
tracing::info!(
app = APP,
config = %config.path().display(),
http = %server.http_addr,
grpc = %server.grpc_addr,
assets = assets::count(),
email = email_enabled,
email_queue_depth,
event_forwarding,
"oauth serving"
);
Ok(Runtime {
service,
http,
grpc,
grpc_listener,
})
}
pub struct Runtime {
service: AuthSvc,
http: actix_web::dev::Server,
grpc: crate::server::grpc::GrpcRouter,
grpc_listener: tokio::net::TcpListener,
}
impl Runtime {
pub fn service(&self) -> &AuthSvc {
&self.service
}
pub async fn serve_until(self, shutdown: tokio_util::sync::CancellationToken) -> Result<()> {
let Self {
service,
http,
grpc,
grpc_listener,
} = self;
let handle = http.handle();
let grpc_cancel = shutdown.clone();
let incoming = tokio_stream::wrappers::TcpListenerStream::new(grpc_listener);
let grpc = grpc.serve_with_incoming_shutdown(incoming, async move {
grpc_cancel.cancelled().await;
});
tokio::pin!(http, grpc);
let mut grpc_done = false;
let mut http_done = false;
let result = tokio::select! {
result = &mut http => { http_done = true; result.context("identity HTTP server stopped") },
result = &mut grpc => { grpc_done = true; result.context("identity gRPC server stopped") },
_ = shutdown.cancelled() => Ok(()),
};
shutdown.cancel();
let http_shutdown = async {
if !http_done {
let stop = handle.stop(true);
let (_, result) = tokio::join!(stop, &mut http);
if let Err(error) = result {
tracing::warn!(%error, "identity HTTP shutdown failed");
}
}
};
let grpc_shutdown = async {
if !grpc_done {
match tokio::time::timeout(
std::time::Duration::from_secs(GRPC_SHUTDOWN_GRACE_SECS),
&mut grpc,
)
.await
{
Ok(Ok(())) => {}
Ok(Err(error)) => tracing::warn!(%error, "identity gRPC shutdown failed"),
Err(_) => tracing::warn!("identity gRPC shutdown timed out"),
}
}
};
tokio::join!(http_shutdown, grpc_shutdown);
if let Some(email) = service.email()
&& let Err(error) = email.shutdown(EMAIL_SHUTDOWN_GRACE).await
{
tracing::warn!(%error, "draining the email queue failed");
}
if let Some(events) = service.events()
&& let Err(error) = events.shutdown().await
{
tracing::warn!(%error, "shutting down identity producer failed");
}
result
}
}