#![deny(missing_docs)]
#![warn(unreachable_pub)]
#![allow(clippy::missing_errors_doc)]
mod async_sleep;
mod authz;
mod bootstrap_config;
mod common;
mod context_data_api;
mod entity_builder;
mod http;
mod http_utils;
mod init;
mod jwt;
mod lock;
mod log;
#[doc(hidden)]
pub mod sparkv;
#[cfg(not(target_arch = "wasm32"))]
#[cfg(feature = "blocking")]
pub mod blocking;
#[doc(hidden)]
#[cfg(test)]
mod tests;
use std::collections::{HashMap, HashSet};
use std::{fmt::Write, sync::Arc};
use crate::authz::metrics::MetricsCollector;
use crate::context_data_api::DataStore;
pub use crate::context_data_api::{
CedarType, CedarValueMapper, ConfigValidationError, DataApi, DataEntry, DataError,
DataStoreConfig, DataStoreStats, DataValidator, ExtensionValue, ValidationConfig,
ValidationError, ValidationResult, ValueMappingError,
};
pub use crate::jwt::TrustedIssuerLoadingInfo;
use authz::Authz;
pub use authz::request::{
AuthorizeMultiIssuerRequest, BatchAuthorizeMultiIssuerRequest, BatchAuthorizeResponse,
BatchAuthorizeUnsignedRequest, BatchItem, CedarEntityMapping, EntityData, RequestUnsigned,
TokenInput,
};
pub use authz::{AuthorizeError, AuthorizeResult, BatchItemError, MultiIssuerAuthorizeResult};
pub use bootstrap_config::*;
pub use cedar_policy::PolicyId;
use common::app_types::{self, ApplicationName};
#[cfg(feature = "tools")]
pub use common::policy_store::validate::{
Diagnostic, LevelResult, ValidateInfraError, ValidationReport,
};
pub use common::policy_store::{PolicyEffect, PolicyMetadata};
pub use http::HttpClientConfig;
use init::ServiceFactory;
use init::policy_store::{LoadedPolicyStore, load_policy_store};
use init::policy_store_refresh::{
AuthzRebuilder, PolicyStoreRefreshHandle, RefreshSource, RefreshWorkerSeed, WorkerContext,
spawn_refresh_worker,
};
use init::service_config::{ServiceConfig, ServiceConfigError};
use init::service_factory::ServiceInitError;
use lock::InitLockServiceError;
use lock::health_registry::HealthStatus;
use log::interface::LogWriter;
use log::{BaseLogEntry, LogEntry};
pub use log::{LogLevel, LogStorage};
use semver::Version;
const BUILD_COMMIT: Option<&str> = option_env!("CEDARLING_BUILD_COMMIT");
const BUILD_TIMESTAMP: Option<&str> = option_env!("CEDARLING_BUILD_TIMESTAMP");
#[doc(hidden)]
pub mod bindings {
pub use cedar_policy;
pub use super::log::{
AuthorizationLogInfo, Decision, Diagnostics, LogEntry, PolicyEvaluationError,
};
pub use crate::http::spawn_task;
pub use crate::sparkv;
pub use serde_json;
pub use serde_yaml_ng;
}
#[derive(Debug, thiserror::Error)]
pub enum InitCedarlingError {
#[error(transparent)]
ServiceConfig(#[from] ServiceConfigError),
#[error(transparent)]
ServiceInit(#[from] ServiceInitError),
#[error(transparent)]
BootstrapConfigLoading(#[from] BootstrapConfigLoadingError),
#[error(transparent)]
DataStoreInit(#[from] ConfigValidationError),
#[cfg(feature = "blocking")]
#[error(transparent)]
RuntimeInit(std::io::Error),
#[error("failed to initialize the Lock Service: {0}")]
InitLockService(#[from] InitLockServiceError),
}
#[derive(Clone)]
pub struct Cedarling {
log: log::Logger,
authz: Arc<arc_swap::ArcSwap<Authz>>,
data: Arc<DataStore>,
_refresh_handle: Option<Arc<PolicyStoreRefreshHandle>>,
}
impl Cedarling {
#[cfg(not(target_arch = "wasm32"))]
pub async fn new_with_env(
raw_config: Option<BootstrapConfigRaw>,
) -> Result<Cedarling, InitCedarlingError> {
let config = BootstrapConfig::from_raw_config_and_env(raw_config)?;
Self::new(&config).await
}
pub async fn new(config: &BootstrapConfig) -> Result<Cedarling, InitCedarlingError> {
let pdp_id = app_types::PdpID::new();
let app_name = (!config.application_name.is_empty())
.then(|| ApplicationName::from(config.application_name.clone()));
let metrics = Arc::new(
if config
.lock_config
.as_ref()
.is_some_and(|c| c.telemetry_interval.is_some())
{
MetricsCollector::new(0)
} else {
MetricsCollector::disabled()
},
);
let log = crate::log::init_logger(
&config.log_config,
pdp_id,
app_name,
config.lock_config.as_ref(),
metrics.clone(),
config.http_client_config,
)
.await?;
log.log_any(
LogEntry::new(BaseLogEntry::new_system_opt_request_id(
LogLevel::INFO,
None,
))
.set_message("Cedarling initialization started".to_string())
.set_build_info(BUILD_COMMIT, BUILD_TIMESTAMP),
);
let (service_config, refresh_seed) = perform_bootstrap_load(config, &log).await?;
let policy_count = service_config
.policy_store
.policies
.get_set()
.num_of_policies();
metrics.set_policy_count(policy_count);
let data = Arc::new(DataStore::new(
config.data_store_config.clone(),
metrics.clone(),
)?);
let mut service_factory = ServiceFactory::new(
config,
service_config,
log.clone(),
data.clone(),
metrics.clone(),
);
if let Some(metadata) = service_factory.policy_store_metadata() {
log_policy_store_metadata(&log, metadata);
}
if let Some(registry) = log.health_registry() {
registry.register("core", || HealthStatus::Success);
registry.register("policy_load", move || {
if policy_count > 0 {
HealthStatus::Success
} else {
HealthStatus::Failure
}
});
}
let authz = service_factory.authz_service().await?;
let authz_swap = Arc::new(arc_swap::ArcSwap::from(authz));
let refresh_handle = maybe_spawn_refresh_worker(
config,
&service_factory,
authz_swap.clone(),
log.clone(),
data.clone(),
metrics.clone(),
refresh_seed,
);
Ok(Cedarling {
log,
authz: authz_swap,
data,
_refresh_handle: refresh_handle,
})
}
#[allow(unknown_lints, clippy::unused_async, clippy::unused_async_trait_impl)]
pub async fn authorize_unsigned(
&self,
request: RequestUnsigned,
) -> Result<AuthorizeResult, AuthorizeError> {
self.authz.load().authorize_unsigned(&request)
}
#[allow(unknown_lints, clippy::unused_async, clippy::unused_async_trait_impl)]
pub async fn authorize_unsigned_batch(
&self,
request: BatchAuthorizeUnsignedRequest,
) -> Result<BatchAuthorizeResponse<Result<AuthorizeResult, BatchItemError>>, AuthorizeError>
{
self.authz.load().authorize_unsigned_batch(&request)
}
#[allow(unknown_lints, clippy::unused_async, clippy::unused_async_trait_impl)]
pub async fn authorize_multi_issuer(
&self,
request: AuthorizeMultiIssuerRequest,
) -> Result<MultiIssuerAuthorizeResult, AuthorizeError> {
self.authz.load().authorize_multi_issuer(&request)
}
#[allow(unknown_lints, clippy::unused_async, clippy::unused_async_trait_impl)]
pub async fn authorize_multi_issuer_batch(
&self,
request: BatchAuthorizeMultiIssuerRequest,
) -> Result<
BatchAuthorizeResponse<Result<MultiIssuerAuthorizeResult, BatchItemError>>,
AuthorizeError,
> {
self.authz.load().authorize_multi_issuer_batch(&request)
}
pub fn get_matching_policies_unsigned(
&self,
principal: Option<&EntityData>,
actions: &[String],
resources: &[EntityData],
) -> Result<Vec<PolicyMetadata>, AuthorizeError> {
self.authz
.load()
.get_matching_policies_unsigned(principal, actions, resources)
}
pub fn get_matching_policies_multi_issuer(
&self,
tokens: &[TokenInput],
actions: &[String],
resources: &[EntityData],
) -> Result<Vec<PolicyMetadata>, AuthorizeError> {
self.authz
.load()
.get_matching_policies_multi_issuer(tokens, actions, resources)
}
pub fn annotations_map<'a>(
&self,
ids: impl IntoIterator<Item = &'a PolicyId>,
) -> HashMap<String, String> {
self.authz.load().annotations_map(ids)
}
pub fn annotation_values<'a>(
&self,
ids: impl IntoIterator<Item = &'a PolicyId>,
key: &str,
) -> Vec<String> {
self.authz.load().annotation_values(ids, key)
}
pub fn annotations_by_policy<'a>(
&self,
ids: impl IntoIterator<Item = &'a PolicyId>,
) -> HashMap<String, HashMap<String, String>> {
self.authz.load().annotations_by_policy(ids)
}
pub async fn shut_down(&self) {
self.log.shut_down().await;
}
}
impl TrustedIssuerLoadingInfo for Cedarling {
fn is_trusted_issuer_loaded_by_name(&self, issuer_id: &str) -> bool {
self.authz
.load()
.is_trusted_issuer_loaded_by_name(issuer_id)
}
fn is_trusted_issuer_loaded_by_iss(&self, iss_claim: &str) -> bool {
self.authz.load().is_trusted_issuer_loaded_by_iss(iss_claim)
}
fn total_issuers(&self) -> usize {
self.authz.load().total_issuers()
}
fn loaded_trusted_issuers_count(&self) -> usize {
self.authz.load().loaded_trusted_issuers_count()
}
fn loaded_trusted_issuer_ids(&self) -> HashSet<String> {
self.authz.load().loaded_trusted_issuer_ids()
}
fn failed_trusted_issuer_ids(&self) -> HashSet<String> {
self.authz.load().failed_trusted_issuer_ids()
}
}
async fn perform_bootstrap_load(
config: &BootstrapConfig,
log: &log::Logger,
) -> Result<(ServiceConfig, RefreshWorkerSeed), ServiceConfigError> {
let raw_load: Result<(http::HttpClient, LoadedPolicyStore), ServiceConfigError> = async {
let http_client = http::HttpClient::new(config.http_client_config)?;
let loaded = load_policy_store(
&config.policy_store_config,
&http_client,
config.authorization_config.strict_schema_validation,
)
.await?;
Ok((http_client, loaded))
}
.await;
let (http_client, loaded) = raw_load
.inspect(|_| {
log.log_any(
LogEntry::new(BaseLogEntry::new_system_opt_request_id(
LogLevel::DEBUG,
None,
))
.set_message("configuration parsed successfully".to_string()),
);
})
.inspect_err(|err| {
log.log_any(
LogEntry::new(BaseLogEntry::new_system_opt_request_id(
LogLevel::ERROR,
None,
))
.set_error(err.to_string())
.set_message("configuration parsed with error".to_string()),
);
})?;
let LoadedPolicyStore {
store: policy_store,
body_hash,
validators,
} = loaded;
Ok((
ServiceConfig {
policy_store,
http_client,
},
RefreshWorkerSeed {
initial_body_hash: body_hash,
initial_validators: validators,
},
))
}
fn maybe_spawn_refresh_worker(
config: &BootstrapConfig,
service_factory: &ServiceFactory<'_>,
authz_swap: Arc<arc_swap::ArcSwap<authz::Authz>>,
log: log::Logger,
data: Arc<context_data_api::DataStore>,
metrics: Arc<authz::metrics::MetricsCollector>,
seed: RefreshWorkerSeed,
) -> Option<Arc<PolicyStoreRefreshHandle>> {
if !config.policy_store_config.refresh_enabled() {
return None;
}
let source = RefreshSource::from_policy_store_source(&config.policy_store_config.source)?;
let (interval_secs, clamped) = config.policy_store_config.effective_refresh_interval();
if clamped {
log.log_any(
LogEntry::new(BaseLogEntry::new_system_opt_request_id(LogLevel::WARN, None))
.set_message(format!(
"CEDARLING_POLICY_STORE_REFRESH_INTERVAL={} is below the minimum; clamped to {} seconds",
config.policy_store_config.refresh_interval_secs,
interval_secs,
)),
);
}
let rebuilder = AuthzRebuilder {
jwt_config: config.jwt_config.clone(),
authorization_config: config.authorization_config.clone(),
http_client: service_factory.http_client_for_refresh(),
log: log.clone(),
data_store: data,
metrics: metrics.clone(),
};
let ctx = WorkerContext {
source,
interval_secs,
http_client: service_factory.http_client_for_refresh(),
rebuilder,
authz_swap,
metrics,
log,
initial_body_hash: seed.initial_body_hash,
initial_validators: seed.initial_validators,
strict_schema_validation: config.authorization_config.strict_schema_validation,
};
Some(Arc::new(spawn_refresh_worker(ctx)))
}
fn log_policy_store_metadata(
log: &log::Logger,
metadata: &crate::common::policy_store::PolicyStoreMetadata,
) {
let mut details = format!(
"Policy store '{}' (ID: {}) v{} loaded",
metadata.name(),
if metadata.id().is_empty() {
"<auto>"
} else {
metadata.id()
},
metadata.version()
);
if let Some(desc) = metadata.description() {
let _ = write!(details, " - {desc}");
}
let _ = write!(details, " [Cedar {}]", metadata.cedar_version());
if let Some(created) = metadata.created_date() {
let _ = write!(details, " (created: {})", created.format("%Y-%m-%d"));
}
if let Some(updated) = metadata.updated_date() {
let _ = write!(details, " (updated: {})", updated.format("%Y-%m-%d"));
}
log.log_any(
LogEntry::new(BaseLogEntry::new_system_opt_request_id(
LogLevel::DEBUG,
None,
))
.set_message(details),
);
let current_cedar_version: Version = cedar_policy::get_lang_version();
match metadata.is_compatible_with_cedar(¤t_cedar_version) {
Ok(true) => {
log.log_any(
LogEntry::new(BaseLogEntry::new_system_opt_request_id(
LogLevel::DEBUG,
None,
))
.set_message(format!(
"Policy store Cedar version {} is compatible with runtime version {}",
metadata.cedar_version(),
current_cedar_version
)),
);
},
Ok(false) => {
log.log_any(
LogEntry::new(BaseLogEntry::new_system_opt_request_id(
LogLevel::WARN,
None,
))
.set_message(format!(
"Policy store Cedar version {} may not be compatible with runtime version {}",
metadata.cedar_version(),
current_cedar_version
)),
);
},
Err(e) => {
log.log_any(
LogEntry::new(BaseLogEntry::new_system_opt_request_id(
LogLevel::WARN,
None,
))
.set_message(format!("Could not check Cedar version compatibility: {e}")),
);
},
}
if let Some(parsed_version) = metadata.version_parsed() {
log.log_any(
LogEntry::new(BaseLogEntry::new_system_opt_request_id(
LogLevel::TRACE,
None,
))
.set_message(format!(
"Policy store semantic version: {}.{}.{}",
parsed_version.major, parsed_version.minor, parsed_version.patch
)),
);
}
}
impl LogStorage for Cedarling {
fn pop_logs(&self) -> Vec<serde_json::Value> {
self.log.pop_logs()
}
fn get_log_by_id(&self, id: &str) -> Option<serde_json::Value> {
self.log.get_log_by_id(id)
}
fn get_log_ids(&self) -> Vec<String> {
self.log.get_log_ids()
}
fn get_logs_by_tag(&self, tag: &str) -> Vec<serde_json::Value> {
self.log.get_logs_by_tag(tag)
}
fn get_logs_by_request_id(&self, request_id: &str) -> Vec<serde_json::Value> {
self.log.get_logs_by_request_id(request_id)
}
fn get_logs_by_request_id_and_tag(&self, id: &str, tag: &str) -> Vec<serde_json::Value> {
self.log.get_logs_by_request_id_and_tag(id, tag)
}
}
fn calculate_capacity_usage(
entry_count: usize,
max_entries: usize,
memory_alert_threshold: f64,
) -> (f64, bool) {
#[allow(clippy::cast_precision_loss)]
let capacity_usage_percent = if max_entries > 0 {
(entry_count as f64 / max_entries as f64) * 100.0
} else {
0.0 };
let memory_alert_triggered = capacity_usage_percent >= memory_alert_threshold;
(capacity_usage_percent, memory_alert_triggered)
}
impl DataApi for Cedarling {
fn push_data_ctx(
&self,
key: &str,
value: serde_json::Value,
ttl: Option<std::time::Duration>,
) -> Result<(), DataError> {
self.data.push(key, value, ttl)?;
let config = self.data.config();
if config.max_entries > 0 {
let entry_count = self.data.count();
let (capacity_usage_percent, memory_alert_triggered) = calculate_capacity_usage(
entry_count,
config.max_entries,
config.memory_alert_threshold,
);
if memory_alert_triggered {
let log_entry = LogEntry::new(BaseLogEntry::new_system_opt_request_id(
LogLevel::WARN,
None,
))
.set_message(format!(
"DataStore memory usage alert: {:.1}% capacity used ({}/{} entries), threshold: {:.1}%",
capacity_usage_percent,
entry_count,
config.max_entries,
config.memory_alert_threshold
));
self.log.log_any(log_entry);
}
}
Ok(())
}
fn get_data_ctx(&self, key: &str) -> Result<Option<serde_json::Value>, DataError> {
Ok(self.data.get(key))
}
fn get_data_entry_ctx(&self, key: &str) -> Result<Option<DataEntry>, DataError> {
Ok(self.data.get_entry(key))
}
fn remove_data_ctx(&self, key: &str) -> Result<bool, DataError> {
Ok(self.data.remove(key))
}
fn clear_data_ctx(&self) -> Result<(), DataError> {
self.data.clear();
Ok(())
}
fn list_data_ctx(&self) -> Result<Vec<DataEntry>, DataError> {
Ok(self.data.list_entries())
}
fn get_stats_ctx(&self) -> Result<DataStoreStats, DataError> {
let config = self.data.config();
let entry_count = self.data.count();
let total_size_bytes = self.data.total_size();
let avg_entry_size_bytes = total_size_bytes.checked_div(entry_count).unwrap_or(0);
let (capacity_usage_percent, memory_alert_triggered) = calculate_capacity_usage(
entry_count,
config.max_entries,
config.memory_alert_threshold,
);
Ok(DataStoreStats {
entry_count,
max_entries: config.max_entries,
max_entry_size: config.max_entry_size,
metrics_enabled: config.enable_metrics,
total_size_bytes,
avg_entry_size_bytes,
capacity_usage_percent,
memory_alert_threshold: config.memory_alert_threshold,
memory_alert_triggered,
})
}
}
#[cfg(feature = "tools")]
impl Cedarling {
#[must_use]
pub fn all_policy_metadata(&self) -> Vec<PolicyMetadata> {
self.authz.load().all_policy_metadata()
}
#[allow(clippy::too_many_lines)]
pub async fn validate_policy_store(
config: &PolicyStoreConfig,
http_config: &crate::http::HttpClientConfig,
) -> Result<ValidationReport, ValidateInfraError> {
let http_client = crate::http::HttpClient::new(*http_config)?;
let load_result =
crate::init::policy_store::load_policy_store(config, &http_client, false).await;
match load_result {
Err(e) => {
use crate::init::policy_store::PolicyStoreLoadError;
let err_str = e.to_string();
let is_metadata = matches!(&e, PolicyStoreLoadError::Validation(_));
let is_parse = matches!(
&e,
PolicyStoreLoadError::ParseJson(_)
| PolicyStoreLoadError::ParseYaml(_)
| PolicyStoreLoadError::Conversion(_)
| PolicyStoreLoadError::InvalidStore(_)
);
match e {
PolicyStoreLoadError::FetchFromLockServer(_)
| PolicyStoreLoadError::Archive(_)
| PolicyStoreLoadError::Directory(_) => {
Err(ValidateInfraError::Io(std::io::Error::other(err_str)))
},
PolicyStoreLoadError::ParseFile(_, io_err) => {
Err(ValidateInfraError::Io(io_err))
},
_ if is_metadata => {
let diag = Diagnostic {
file: "<policy-store>".into(),
line: None,
column: None,
message: err_str,
};
Ok(ValidationReport {
parse: LevelResult::Skipped {
reason: "metadata check failed".into(),
},
schema: LevelResult::Skipped {
reason: "metadata check failed".into(),
},
metadata: LevelResult::Failed { errors: vec![diag] },
})
},
_ if is_parse => {
let diag = Diagnostic {
file: "<policy-store>".into(),
line: None,
column: None,
message: err_str,
};
Ok(ValidationReport {
parse: LevelResult::Failed { errors: vec![diag] },
schema: LevelResult::Skipped {
reason: "parse failed".into(),
},
metadata: LevelResult::Skipped {
reason: "parse failed".into(),
},
})
},
_ => Err(ValidateInfraError::Io(std::io::Error::other(err_str))),
}
},
Ok(loaded) => {
let schema_res = if let Some(schema) = &loaded.store.store.schema {
let validator = cedar_policy::Validator::new(schema.schema.clone());
let result = validator.validate(
loaded.store.store.policies.get_set(),
cedar_policy::ValidationMode::Strict,
);
if result.validation_passed() {
LevelResult::Ok
} else {
let errors = result
.validation_errors()
.map(|e| Diagnostic {
file: e.policy_id().to_string(),
line: None,
column: None,
message: e.to_string(),
})
.collect();
LevelResult::Failed { errors }
}
} else {
LevelResult::Skipped {
reason: "no schema present".into(),
}
};
let metadata_res = match &loaded.store.metadata {
Some(metadata) => {
use crate::common::policy_store::validator::MetadataValidator;
match MetadataValidator::validate(metadata) {
Ok(()) => LevelResult::Ok,
Err(e) => LevelResult::Failed {
errors: vec![Diagnostic {
file: "<metadata>".into(),
line: None,
column: None,
message: e.to_string(),
}],
},
}
},
None => {
match crate::common::policy_store::validator::validate_legacy_metadata(
&loaded.store.store,
) {
Ok(()) => LevelResult::Ok,
Err(e) => LevelResult::Failed {
errors: vec![Diagnostic {
file: "<inline>".into(),
line: None,
column: None,
message: e.to_string(),
}],
},
}
}
};
Ok(ValidationReport {
parse: LevelResult::Ok,
schema: schema_res,
metadata: metadata_res,
})
},
}
}
}