mod apikey;
mod audit_export;
mod authn;
mod authz;
mod control_plane;
pub(crate) mod events;
mod idp;
mod mappings;
pub(crate) mod readiness;
pub use readiness::auth_readiness_triples;
#[cfg(test)]
mod tests;
use std::sync::Arc;
pub use crate::proto::udb::core::apikey::services::v1::api_key_service_server::ApiKeyServiceServer;
pub use crate::proto::udb::core::authn::services::v1::authn_service_server::AuthnServiceServer;
pub use crate::proto::udb::core::authz::services::v1::authz_service_server::AuthzServiceServer;
pub use idp::IdentityProviderServiceServer;
pub(crate) use idp::spawn_scim_http_from_env;
pub use control_plane::ControlPlaneServiceServer;
pub use apikey::ApiKeyServiceImpl;
pub use authn::AuthnServiceImpl;
pub use authz::AuthzServiceImpl;
#[cfg(feature = "redis")]
use crate::runtime::authn::RedisSessionStore;
use crate::runtime::authn::{
AccountStatus, AuthnConfig, PostgresApiKeyStore, PostgresSessionStore, PostgresUserStore,
SessionStore, UserStore,
};
use crate::runtime::authz::AuthzSnapshot;
use crate::runtime::security::SecurityConfig;
use super::DataBrokerService;
pub(super) fn now_unix() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
impl DataBrokerService {
pub(crate) fn build_auth_services(
&self,
) -> (AuthnServiceImpl, AuthzServiceImpl, ApiKeyServiceImpl) {
let policies = self
.abac_policies
.read()
.map(|guard| guard.clone())
.unwrap_or_default();
let mut snapshot = AuthzSnapshot::from_abac_policies("live-abac", &policies);
snapshot.default_allow = self.abac_default_allow;
let shared_snapshot = Arc::new(arc_swap::ArcSwap::from_pointee(snapshot));
let runtime = self.runtime.load_full();
let pg_pool = runtime
.native_store_pool_for_service("authn", true, "")
.ok();
let authn_config = AuthnConfig::from_env();
let security = SecurityConfig::current();
let event_sink: Arc<dyn events::AuthEventSink> = match pg_pool.clone() {
Some(pool) => Arc::new(
events::OutboxAuthEventSink::new(
pool.clone(),
runtime.config().cdc.outbox_relation(),
)
.with_exports(audit_export::export_sinks_from_env(Some(&pool)))
.with_metrics(self.metrics.clone()),
),
None => events::noop_sink(),
};
let (authn_service, api_key_service) = if let Some(pool) = pg_pool.clone() {
#[cfg(feature = "redis")]
let jti_denylist = runtime.redis_clone().map(|redis| {
crate::runtime::authn::revocation::JtiDenylist::new(
redis,
security.jwt_access_ttl_secs,
)
});
let api_key_store = Arc::new(PostgresApiKeyStore::new(pool.clone(), ""));
let session_store: Arc<dyn SessionStore> =
if authn_config.session_backend.eq_ignore_ascii_case("redis") {
#[cfg(feature = "redis")]
{
if let Some(redis) = runtime.redis_clone() {
Arc::new(RedisSessionStore::new(
redis,
"udb:authn",
authn_config.session_ttl_secs,
))
} else {
Arc::new(PostgresSessionStore::new(pool.clone(), ""))
}
}
#[cfg(not(feature = "redis"))]
{
Arc::new(PostgresSessionStore::new(pool.clone(), ""))
}
} else {
Arc::new(PostgresSessionStore::new(pool.clone(), ""))
};
let authn = AuthnServiceImpl::with_stores(
authn_config.clone(),
security,
session_store,
api_key_store.clone(),
Arc::new(PostgresUserStore::new(pool.clone(), "")),
)
.with_postgres(Some(pool.clone()))
.with_runtime(Some(runtime.clone()))
.with_event_sink(event_sink.clone())
.with_metrics(self.metrics.clone())
.with_authz_snapshot(Some(shared_snapshot.clone()));
#[cfg(feature = "redis")]
let authn = authn.with_jti_denylist(jti_denylist);
(
authn,
ApiKeyServiceImpl::with_store(authn_config, api_key_store)
.with_postgres(Some(pool.clone()))
.with_runtime(Some(runtime.clone()))
.with_event_sink(event_sink.clone()),
)
} else {
(
AuthnServiceImpl::new(authn_config.clone(), security)
.with_event_sink(event_sink.clone())
.with_metrics(self.metrics.clone())
.with_authz_snapshot(Some(shared_snapshot.clone())),
ApiKeyServiceImpl::new(authn_config).with_event_sink(event_sink.clone()),
)
};
(
authn_service,
AuthzServiceImpl::shared(shared_snapshot)
.with_postgres(pg_pool)
.with_runtime(Some(runtime.clone()))
.with_event_sink(event_sink)
.with_metrics(self.metrics.clone())
.with_channels(Some(runtime.channels().clone())),
api_key_service,
)
}
}
pub async fn bootstrap_admin_user(
dsn: &str,
username: &str,
email: &str,
password: &str,
tenant: &str,
project: &str,
) -> Result<BootstrapAdmin, String> {
use crate::proto::udb::core::authn::entity::v1 as authn_entity_pb;
use crate::proto::udb::core::authn::services::v1 as authn_pb;
use crate::proto::udb::core::authn::services::v1::authn_service_server::AuthnService;
use tonic::Request;
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(4)
.acquire_timeout(std::time::Duration::from_secs(10))
.connect(dsn)
.await
.map_err(|err| format!("connect postgres '{dsn}': {err}"))?;
for stmt in crate::runtime::native_catalog::native_service_catalog_ddl() {
if let Err(err) = sqlx::raw_sql(&stmt).execute(&pool).await {
tracing::debug!(error = %err, "bootstrap schema-ensure statement skipped");
}
}
let tenant_uuid = ensure_tenant(&pool, tenant, None).await?;
let session_store: Arc<dyn SessionStore> =
Arc::new(PostgresSessionStore::new(pool.clone(), ""));
let user_store: Arc<dyn UserStore> = Arc::new(PostgresUserStore::new(pool.clone(), ""));
let svc = AuthnServiceImpl::with_stores(
AuthnConfig::from_env(),
SecurityConfig::current(),
session_store,
Arc::new(PostgresApiKeyStore::new(pool.clone(), "")),
user_store.clone(),
)
.with_postgres(Some(pool.clone()));
let login_name = username.trim().to_ascii_lowercase();
let (user_id, needs_activation) = match user_store
.get_user_by_username(&login_name)
.await
.map_err(|err| format!("lookup user failed: {err}"))?
{
Some(existing) => (existing.user_id, existing.status != AccountStatus::Active),
None => {
let created = svc
.create_user(Request::new(authn_pb::CreateUserRequest {
username: username.to_string(),
email: email.to_string(),
password: password.to_string(),
tenant_id: tenant_uuid.clone(),
project_id: project.to_string(),
full_name: "Bootstrap Admin".to_string(),
..Default::default()
}))
.await
.map_err(|err| format!("create_user failed: {err}"))?
.into_inner();
let new_id = created
.user
.ok_or_else(|| "create_user returned no user".to_string())?
.user_id;
(new_id, true)
}
};
if needs_activation {
svc.change_user_status(Request::new(authn_pb::ChangeUserStatusRequest {
user_id: user_id.clone(),
new_status: authn_entity_pb::UserStatus::Active as i32,
reason: "offline bootstrap".to_string(),
..Default::default()
}))
.await
.map_err(|err| format!("activate user failed: {err}"))?;
}
seed_system_authz_defaults(&pool)
.await
.map_err(|err| format!("seed system authz defaults failed: {err}"))?;
{
use crate::runtime::native_catalog::native_model;
let ur = native_model(
"udb.core.authz.entity.v1.UserRole",
&[
"user_role_id",
"user_id",
"role_id",
"domain",
"assigned_by",
"tenant_id",
"created_by",
],
);
sqlx::query(&format!(
"INSERT INTO {rel} \
({user_role_id}, {user_id}, {role_id}, {domain}, {assigned_by}, {tenant_id}, {created_by}) \
VALUES (gen_random_uuid(), $1::UUID, $2::UUID, '', $1::UUID, $3, $1::UUID) \
ON CONFLICT ({user_id}, {role_id}, {domain}) DO UPDATE SET \
{assigned_by} = EXCLUDED.{assigned_by}, {tenant_id} = EXCLUDED.{tenant_id}, {created_by} = EXCLUDED.{created_by}",
rel = ur.relation,
user_role_id = ur.q("user_role_id"),
user_id = ur.q("user_id"),
role_id = ur.q("role_id"),
domain = ur.q("domain"),
assigned_by = ur.q("assigned_by"),
tenant_id = ur.q("tenant_id"),
created_by = ur.q("created_by"),
))
.bind(&user_id)
.bind(SYSTEM_ORG_OWNER_ROLE_ID)
.bind(&tenant_uuid)
.execute(&pool)
.await
.map_err(|err| format!("assign organization_owner role failed: {err}"))?;
}
Ok(BootstrapAdmin {
user_id,
tenant_id: tenant_uuid,
})
}
pub struct BootstrapAdmin {
pub user_id: String,
pub tenant_id: String,
}
pub(crate) const SYSTEM_ORG_OWNER_ROLE_ID: &str = "00000000-0000-0000-0000-0000000a0001";
pub(crate) const SYSTEM_ORG_OWNER_POLICY_ID: &str = "00000000-0000-0000-0000-0000000a0002";
pub(crate) const ORG_OWNER_ROLE_CODE: &str = "organization_owner";
pub(crate) const DEFAULT_TENANT_CODE: &str = "acme";
pub(crate) const DEFAULT_TENANT_ID: &str = "00000000-0000-0000-0000-0000000d0001";
pub(crate) fn default_tenant_code() -> String {
std::env::var("UDB_DEFAULT_TENANT_CODE")
.ok()
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| DEFAULT_TENANT_CODE.to_string())
}
pub(crate) fn default_tenant_id() -> String {
std::env::var("UDB_DEFAULT_TENANT_ID")
.ok()
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| DEFAULT_TENANT_ID.to_string())
}
pub(crate) async fn ensure_tenant(
pool: &sqlx::PgPool,
tenant_in: &str,
name: Option<&str>,
) -> Result<String, String> {
use crate::runtime::native_catalog::native_model;
let trimmed = tenant_in.trim();
if trimmed.is_empty() {
return Err("tenant code or id is required".to_string());
}
let is_uuid = uuid::Uuid::parse_str(trimmed).is_ok();
let m = native_model(
"udb.core.tenant.entity.v1.Tenant",
&[
"tenant_id",
"code",
"name",
"type",
"status",
"config",
"branding",
"audit_info",
"deleted_at",
],
);
let lookup_sql = if is_uuid {
format!(
"SELECT {tid}::text FROM {rel} WHERE {tid} = $1::UUID AND {del} IS NULL",
rel = m.relation,
tid = m.q("tenant_id"),
del = m.q("deleted_at"),
)
} else {
format!(
"SELECT {tid}::text FROM {rel} WHERE {code} = $1 AND {del} IS NULL",
rel = m.relation,
tid = m.q("tenant_id"),
code = m.q("code"),
del = m.q("deleted_at"),
)
};
if let Some(found) = sqlx::query_scalar::<_, String>(&lookup_sql)
.bind(trimmed)
.fetch_optional(pool)
.await
.map_err(|err| format!("tenant lookup failed: {err}"))?
{
return Ok(found);
}
let default_code = default_tenant_code();
let (canonical_id, code, is_default) = if is_uuid {
(
trimmed.to_string(),
format!("tenant-{}", &trimmed[..8.min(trimmed.len())]),
false,
)
} else if trimmed.eq_ignore_ascii_case(&default_code) {
(default_tenant_id(), trimmed.to_string(), true)
} else {
(uuid::Uuid::new_v4().to_string(), trimmed.to_string(), false)
};
if is_default {
tracing::warn!(
tenant_code = %code,
tenant_id = %canonical_id,
"bootstrapping the well-known DEFAULT tenant — set UDB_DEFAULT_TENANT_CODE or create your own tenant for production"
);
}
let tenant_name = name
.map(str::to_string)
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| format!("{code} (bootstrap)"));
let insert_sql = format!(
"INSERT INTO {rel} ({tid}, {code}, {name}, {type_c}, {status}, {config}, {branding}, {audit}) \
VALUES ($1::UUID, $2, $3, 'ORGANIZATION', 'ACTIVE', '{{}}'::JSONB, '{{}}'::JSONB, '{{}}'::JSONB) \
ON CONFLICT ({code}) DO NOTHING",
rel = m.relation,
tid = m.q("tenant_id"),
code = m.q("code"),
name = m.q("name"),
type_c = m.q("type"),
status = m.q("status"),
config = m.q("config"),
branding = m.q("branding"),
audit = m.q("audit_info"),
);
sqlx::query(&insert_sql)
.bind(&canonical_id)
.bind(&code)
.bind(&tenant_name)
.execute(pool)
.await
.map_err(|err| format!("create tenant '{code}' failed: {err}"))?;
sqlx::query_scalar::<_, String>(&format!(
"SELECT {tid}::text FROM {rel} WHERE {code} = $1 AND {del} IS NULL",
rel = m.relation,
tid = m.q("tenant_id"),
code = m.q("code"),
del = m.q("deleted_at"),
))
.bind(&code)
.fetch_optional(pool)
.await
.map_err(|err| format!("tenant re-lookup failed: {err}"))?
.ok_or_else(|| "tenant ensure failed: row missing after insert".to_string())
}
pub(crate) async fn seed_system_authz_defaults(pool: &sqlx::PgPool) -> Result<(), sqlx::Error> {
use crate::runtime::native_catalog::native_model;
let role = native_model(
"udb.core.authz.entity.v1.Role",
&[
"role_id",
"name",
"description",
"is_system",
"is_active",
"tenant_id",
"project_id",
"role_code",
"scope_type",
],
);
sqlx::query(&format!(
"INSERT INTO {rel} \
({role_id}, {name}, {description}, {is_system}, {is_active}, {tenant_id}, {project_id}, {role_code}, {scope_type}) \
VALUES ($1::UUID, 'Organization Owner', 'Full control across the organization', TRUE, TRUE, '*', '', $2, 'TENANT') \
ON CONFLICT DO NOTHING",
rel = role.relation,
role_id = role.q("role_id"),
name = role.q("name"),
description = role.q("description"),
is_system = role.q("is_system"),
is_active = role.q("is_active"),
tenant_id = role.q("tenant_id"),
project_id = role.q("project_id"),
role_code = role.q("role_code"),
scope_type = role.q("scope_type"),
))
.bind(SYSTEM_ORG_OWNER_ROLE_ID)
.bind(ORG_OWNER_ROLE_CODE)
.execute(pool)
.await?;
let policy = native_model(
"udb.core.authz.entity.v1.PolicyRule",
&[
"policy_id",
"subject",
"domain",
"object",
"action",
"effect",
"condition",
"description",
"is_active",
"tenant_id",
"project_id",
"attributes_json",
],
);
let attributes = serde_json::json!({
"role": ORG_OWNER_ROLE_CODE,
"priority": "0",
"purpose": "",
"relationship": "",
"required_scopes": "",
})
.to_string();
sqlx::query(&format!(
"INSERT INTO {rel} \
({policy_id}, {subject}, {domain}, {object}, {action}, {effect}, {condition}, {description}, {is_active}, {tenant_id}, {project_id}, {attributes_json}) \
VALUES ($1::UUID, '', '*', '*', '*', 'ALLOW', '', 'system: organization_owner full control', TRUE, '*', '', $2::JSONB) \
ON CONFLICT DO NOTHING",
rel = policy.relation,
policy_id = policy.q("policy_id"),
subject = policy.q("subject"),
domain = policy.q("domain"),
object = policy.q("object"),
action = policy.q("action"),
effect = policy.q("effect"),
condition = policy.q("condition"),
description = policy.q("description"),
is_active = policy.q("is_active"),
tenant_id = policy.q("tenant_id"),
project_id = policy.q("project_id"),
attributes_json = policy.q("attributes_json"),
))
.bind(SYSTEM_ORG_OWNER_POLICY_ID)
.bind(attributes)
.execute(pool)
.await?;
Ok(())
}