greentic_runner_host/
lib.rs

1#![deny(unsafe_code)]
2//! Canonical Greentic host runtime.
3//!
4//! This crate owns tenant bindings, pack ingestion/watchers, ingress adapters,
5//! Wasmtime glue, session/state storage, and the HTTP server used by the
6//! `greentic-runner` CLI. Downstream crates embed it either through
7//! [`RunnerConfig`] + [`run`] (HTTP host) or [`HostBuilder`] (direct API access).
8
9use std::collections::{HashMap, HashSet};
10use std::fs;
11use std::path::PathBuf;
12use std::sync::Arc;
13use std::time::Duration;
14
15use crate::secrets::SecretsBackend;
16use anyhow::{Context, Result, anyhow};
17use greentic_config::ResolvedConfig;
18#[cfg(feature = "telemetry")]
19use greentic_config_types::TelemetryExporterKind;
20use greentic_config_types::{
21    NetworkConfig, PackSourceConfig, PacksConfig, PathsConfig, TelemetryConfig,
22};
23#[cfg(feature = "telemetry")]
24use greentic_telemetry::export::{ExportConfig as TelemetryExportConfig, ExportMode, Sampling};
25use runner_core::env::PackConfig;
26use serde_json::json;
27use tokio::signal;
28
29pub mod boot;
30pub mod cache;
31pub mod component_api;
32pub mod config;
33pub mod engine;
34pub mod fault;
35pub mod gtbind;
36pub mod http;
37pub mod operator_metrics;
38pub mod operator_registry;
39pub mod pack;
40pub mod provider;
41pub mod provider_core;
42pub mod provider_core_only;
43pub mod routing;
44pub mod runner;
45pub mod runtime;
46pub mod runtime_wasmtime;
47pub mod secrets;
48pub mod storage;
49pub mod telemetry;
50#[cfg(feature = "fault-injection")]
51pub mod testing;
52pub mod trace;
53pub mod validate;
54pub mod verify;
55pub mod wasi;
56pub mod watcher;
57
58mod activity;
59mod host;
60pub mod oauth;
61
62pub use activity::{Activity, ActivityKind};
63pub use config::HostConfig;
64pub use gtbind::{PackBinding, TenantBindings};
65pub use host::TelemetryCfg;
66pub use host::{HostBuilder, RunnerHost, TenantHandle};
67pub use wasi::{PreopenSpec, RunnerWasiPolicy};
68
69pub use greentic_types::{EnvId, FlowId, PackId, TenantCtx, TenantId};
70
71pub use http::auth::AdminAuth;
72pub use routing::RoutingConfig;
73use routing::TenantRouting;
74pub use runner::HostServer;
75
76#[cfg(test)]
77pub(crate) mod test_support {
78    use super::*;
79    use crate::config::{OperatorPolicy, SecretsPolicy};
80    use crate::runtime::TenantRuntime;
81    use crate::secrets::default_manager;
82    use crate::storage::{new_session_store, new_state_store, session_host_from, state_host_from};
83    use crate::trace::TraceConfig;
84    use crate::validate::ValidationConfig;
85    use tempfile::TempDir;
86
87    pub(crate) fn fixture_pack_path() -> PathBuf {
88        PathBuf::from(env!("CARGO_MANIFEST_DIR"))
89            .join("../../examples/packs/demo.gtpack")
90            .canonicalize()
91            .expect("fixture pack path")
92    }
93
94    fn minimal_config(workspace: &std::path::Path) -> Result<Arc<HostConfig>> {
95        let bindings_path = workspace.join("bindings.yaml");
96        std::fs::write(
97            &bindings_path,
98            r#"
99tenant: demo
100flow_type_bindings: {}
101rate_limits: {}
102retry: {}
103timers: []
104"#,
105        )?;
106        let mut config =
107            HostConfig::load_from_path(&bindings_path).context("load minimal host bindings")?;
108        config.secrets_policy = SecretsPolicy::allow_all();
109        config.operator_policy = OperatorPolicy::allow_all();
110        config.trace = TraceConfig::from_env();
111        config.validation = ValidationConfig::from_env();
112        Ok(Arc::new(config))
113    }
114
115    pub(crate) async fn build_test_runtime() -> Result<(TempDir, Arc<TenantRuntime>)> {
116        let workspace = TempDir::new().context("temp workspace")?;
117        let config = minimal_config(workspace.path())?;
118        let session_store = new_session_store();
119        let session_host = session_host_from(Arc::clone(&session_store));
120        let state_store = new_state_store();
121        let state_host = state_host_from(Arc::clone(&state_store));
122        let secrets = default_manager()?;
123        let pack_path = fixture_pack_path();
124        let runtime = TenantRuntime::load(
125            &pack_path,
126            config,
127            None,
128            Some(&pack_path),
129            None,
130            Arc::new(RunnerWasiPolicy::new()),
131            session_host,
132            Arc::clone(&session_store),
133            Arc::clone(&state_store),
134            state_host,
135            secrets,
136        )
137        .await?;
138        Ok((workspace, runtime))
139    }
140}
141
142/// User-facing configuration for running the unified host.
143#[derive(Clone)]
144pub struct RunnerConfig {
145    pub tenant_bindings: HashMap<String, TenantBindings>,
146    pub pack: PackConfig,
147    pub port: u16,
148    pub refresh_interval: Duration,
149    pub routing: RoutingConfig,
150    pub admin: AdminAuth,
151    pub telemetry: Option<TelemetryCfg>,
152    pub secrets_backend: SecretsBackend,
153    pub wasi_policy: RunnerWasiPolicy,
154    pub resolved_config: ResolvedConfig,
155    pub trace: trace::TraceConfig,
156    pub validation: validate::ValidationConfig,
157}
158
159impl RunnerConfig {
160    /// Build a [`RunnerConfig`] from a resolved greentic-config and the provided binding files.
161    pub fn from_config(resolved_config: ResolvedConfig, bindings: Vec<PathBuf>) -> Result<Self> {
162        if bindings.is_empty() {
163            anyhow::bail!("at least one gtbind file is required");
164        }
165        let tenant_bindings = gtbind::load_gtbinds(&bindings)?;
166        if tenant_bindings.is_empty() {
167            anyhow::bail!("no gtbind files loaded");
168        }
169        let mut pack = pack_config_from(
170            &resolved_config.config.packs,
171            &resolved_config.config.paths,
172            &resolved_config.config.network,
173        )?;
174        maybe_write_gtbind_index(&tenant_bindings, &resolved_config.config.paths, &mut pack)?;
175        let refresh = parse_refresh_interval(std::env::var("PACK_REFRESH_INTERVAL").ok())?;
176        let port = std::env::var("PORT")
177            .ok()
178            .and_then(|value| value.parse().ok())
179            .unwrap_or(8080);
180        let default_tenant = resolved_config
181            .config
182            .dev
183            .as_ref()
184            .map(|dev| dev.default_tenant.clone())
185            .unwrap_or_else(|| "demo".into());
186        let routing = RoutingConfig::from_env_with_default(default_tenant);
187        let paths = &resolved_config.config.paths;
188        ensure_paths_exist(paths)?;
189        let mut wasi_policy = default_wasi_policy(paths);
190        let mut env_allow = HashSet::new();
191        for binding in tenant_bindings.values() {
192            env_allow.extend(binding.env_passthrough.iter().cloned());
193        }
194        for key in env_allow {
195            wasi_policy = wasi_policy.allow_env(key);
196        }
197
198        let admin = AdminAuth::new(resolved_config.config.services.as_ref().and_then(|s| {
199            s.events
200                .as_ref()
201                .and_then(|svc| svc.headers.as_ref())
202                .and_then(|headers| headers.get("x-admin-token").cloned())
203        }));
204        let secrets_backend = SecretsBackend::from_config(&resolved_config.config.secrets)?;
205        Ok(Self {
206            tenant_bindings,
207            pack,
208            port,
209            refresh_interval: refresh,
210            routing,
211            admin,
212            telemetry: telemetry_from(&resolved_config.config.telemetry),
213            secrets_backend,
214            wasi_policy,
215            resolved_config,
216            trace: trace::TraceConfig::from_env(),
217            validation: validate::ValidationConfig::from_env(),
218        })
219    }
220
221    /// Override the HTTP port used by the host server.
222    pub fn with_port(mut self, port: u16) -> Self {
223        self.port = port;
224        self
225    }
226
227    pub fn with_wasi_policy(mut self, policy: RunnerWasiPolicy) -> Self {
228        self.wasi_policy = policy;
229        self
230    }
231}
232
233fn maybe_write_gtbind_index(
234    tenant_bindings: &HashMap<String, TenantBindings>,
235    paths: &PathsConfig,
236    pack: &mut PackConfig,
237) -> Result<()> {
238    let mut uses_locators = false;
239    for binding in tenant_bindings.values() {
240        for pack_binding in &binding.packs {
241            if pack_binding.pack_locator.is_some() {
242                uses_locators = true;
243            }
244        }
245    }
246    if !uses_locators {
247        return Ok(());
248    }
249
250    let mut entries = serde_json::Map::new();
251    for binding in tenant_bindings.values() {
252        let mut packs = Vec::new();
253        for pack_binding in &binding.packs {
254            let locator = pack_binding.pack_locator.as_ref().ok_or_else(|| {
255                anyhow::anyhow!(
256                    "gtbind {} missing pack_locator for pack {}",
257                    binding.tenant,
258                    pack_binding.pack_id
259                )
260            })?;
261            let (name, version_or_digest) =
262                pack_binding.pack_ref.split_once('@').ok_or_else(|| {
263                    anyhow::anyhow!(
264                        "gtbind {} invalid pack_ref {} (expected name@version)",
265                        binding.tenant,
266                        pack_binding.pack_ref
267                    )
268                })?;
269            if name != pack_binding.pack_id {
270                anyhow::bail!(
271                    "gtbind {} pack_ref {} does not match pack_id {}",
272                    binding.tenant,
273                    pack_binding.pack_ref,
274                    pack_binding.pack_id
275                );
276            }
277            let mut entry = serde_json::Map::new();
278            entry.insert("name".to_string(), json!(name));
279            if version_or_digest.contains(':') {
280                entry.insert("digest".to_string(), json!(version_or_digest));
281            } else {
282                entry.insert("version".to_string(), json!(version_or_digest));
283            }
284            entry.insert("locator".to_string(), json!(locator));
285            packs.push(serde_json::Value::Object(entry));
286        }
287        let main_pack = packs
288            .first()
289            .cloned()
290            .ok_or_else(|| anyhow::anyhow!("gtbind {} has no packs", binding.tenant))?;
291        let overlays = packs.into_iter().skip(1).collect::<Vec<_>>();
292        entries.insert(
293            binding.tenant.clone(),
294            json!({
295                "main_pack": main_pack,
296                "overlays": overlays,
297            }),
298        );
299    }
300
301    let index_path = paths.greentic_root.join("packs").join("gtbind.index.json");
302    if let Some(parent) = index_path.parent() {
303        fs::create_dir_all(parent)
304            .with_context(|| format!("failed to create {}", parent.display()))?;
305    }
306    let serialized = serde_json::to_vec_pretty(&serde_json::Value::Object(entries))?;
307    fs::write(&index_path, serialized)
308        .with_context(|| format!("failed to write {}", index_path.display()))?;
309    pack.index_location = runner_core::env::IndexLocation::File(index_path);
310    Ok(())
311}
312
313fn parse_refresh_interval(value: Option<String>) -> Result<Duration> {
314    let raw = value.unwrap_or_else(|| "30s".into());
315    humantime::parse_duration(&raw).map_err(|err| anyhow!("invalid PACK_REFRESH_INTERVAL: {err}"))
316}
317
318fn default_wasi_policy(paths: &PathsConfig) -> RunnerWasiPolicy {
319    let mut policy = RunnerWasiPolicy::default()
320        .with_env("GREENTIC_ROOT", paths.greentic_root.display().to_string())
321        .with_env("GREENTIC_STATE_DIR", paths.state_dir.display().to_string())
322        .with_env("GREENTIC_CACHE_DIR", paths.cache_dir.display().to_string())
323        .with_env("GREENTIC_LOGS_DIR", paths.logs_dir.display().to_string());
324    policy = policy
325        .with_preopen(PreopenSpec::new(&paths.state_dir, "/state"))
326        .with_preopen(PreopenSpec::new(&paths.cache_dir, "/cache"))
327        .with_preopen(PreopenSpec::new(&paths.logs_dir, "/logs"));
328    policy
329}
330
331fn ensure_paths_exist(paths: &PathsConfig) -> Result<()> {
332    for dir in [
333        &paths.greentic_root,
334        &paths.state_dir,
335        &paths.cache_dir,
336        &paths.logs_dir,
337    ] {
338        fs::create_dir_all(dir)
339            .with_context(|| format!("failed to ensure directory {}", dir.display()))?;
340    }
341    Ok(())
342}
343
344fn pack_config_from(
345    packs: &Option<PacksConfig>,
346    paths: &PathsConfig,
347    network: &NetworkConfig,
348) -> Result<PackConfig> {
349    if let Some(cfg) = packs {
350        let cache_dir = cfg.cache_dir.clone();
351        let index_location = match &cfg.source {
352            PackSourceConfig::LocalIndex { path } => {
353                runner_core::env::IndexLocation::File(path.clone())
354            }
355            PackSourceConfig::HttpIndex { url } => {
356                runner_core::env::IndexLocation::from_value(url)?
357            }
358            PackSourceConfig::OciRegistry { reference } => {
359                runner_core::env::IndexLocation::from_value(reference)?
360            }
361        };
362        let public_key = cfg
363            .trust
364            .as_ref()
365            .and_then(|trust| trust.public_keys.first().cloned());
366        return Ok(PackConfig {
367            source: runner_core::env::PackSource::Fs,
368            index_location,
369            cache_dir,
370            public_key,
371            network: Some(network.clone()),
372        });
373    }
374    let mut cfg = PackConfig::default_for_paths(paths)?;
375    cfg.network = Some(network.clone());
376    Ok(cfg)
377}
378
379#[cfg(feature = "telemetry")]
380fn telemetry_from(cfg: &TelemetryConfig) -> Option<TelemetryCfg> {
381    if !cfg.enabled || matches!(cfg.exporter, TelemetryExporterKind::None) {
382        return None;
383    }
384    let mut export = TelemetryExportConfig::json_default();
385    export.mode = match cfg.exporter {
386        TelemetryExporterKind::Otlp => ExportMode::OtlpGrpc,
387        TelemetryExporterKind::Stdout => ExportMode::JsonStdout,
388        TelemetryExporterKind::Gcp => ExportMode::GcpCloudTrace,
389        TelemetryExporterKind::Azure => ExportMode::AzureAppInsights,
390        TelemetryExporterKind::Aws => ExportMode::AwsXRay,
391        TelemetryExporterKind::None => return None,
392    };
393    export.endpoint = cfg.endpoint.clone();
394    export.sampling = Sampling::TraceIdRatio(cfg.sampling as f64);
395    Some(TelemetryCfg {
396        config: greentic_telemetry::TelemetryConfig {
397            service_name: "greentic-runner".into(),
398        },
399        export,
400    })
401}
402
403#[cfg(not(feature = "telemetry"))]
404fn telemetry_from(_cfg: &TelemetryConfig) -> Option<TelemetryCfg> {
405    None
406}
407
408/// Run the unified Greentic runner host until shutdown.
409pub async fn run(cfg: RunnerConfig) -> Result<()> {
410    let RunnerConfig {
411        tenant_bindings,
412        pack,
413        port,
414        refresh_interval,
415        routing,
416        admin,
417        telemetry,
418        secrets_backend,
419        wasi_policy,
420        resolved_config: _resolved_config,
421        trace,
422        validation,
423    } = cfg;
424    #[cfg(not(feature = "telemetry"))]
425    let _ = telemetry;
426
427    let mut builder = HostBuilder::new();
428    for bindings in tenant_bindings.into_values() {
429        let mut host_config = HostConfig::from_gtbind(bindings);
430        host_config.trace = trace.clone();
431        host_config.validation = validation.clone();
432        builder = builder.with_config(host_config);
433    }
434    #[cfg(feature = "telemetry")]
435    if let Some(telemetry) = telemetry.clone() {
436        builder = builder.with_telemetry(telemetry);
437    }
438    builder = builder
439        .with_wasi_policy(wasi_policy.clone())
440        .with_secrets_manager(
441            secrets_backend
442                .build_manager()
443                .context("failed to initialise secrets backend")?,
444        );
445
446    let host = Arc::new(builder.build()?);
447    host.start().await?;
448
449    let (watcher, reload_handle) =
450        watcher::start_pack_watcher(Arc::clone(&host), pack.clone(), refresh_interval).await?;
451
452    let routing = TenantRouting::new(routing.clone());
453    let server = HostServer::new(
454        port,
455        host.active_packs(),
456        routing,
457        host.health_state(),
458        Some(reload_handle),
459        admin.clone(),
460    )?;
461
462    tokio::select! {
463        result = server.serve() => {
464            result?;
465        }
466        _ = signal::ctrl_c() => {
467            tracing::info!("received shutdown signal");
468        }
469    }
470
471    drop(watcher);
472    host.stop().await?;
473    Ok(())
474}
475
476#[cfg(test)]
477mod tests {
478    use super::*;
479    use crate::gtbind::{PackBinding, TenantBindings};
480    use greentic_config_types::PackTrustConfig;
481    use tempfile::TempDir;
482
483    fn paths(temp: &TempDir) -> PathsConfig {
484        PathsConfig {
485            greentic_root: temp.path().join("greentic"),
486            state_dir: temp.path().join("state"),
487            cache_dir: temp.path().join("cache"),
488            logs_dir: temp.path().join("logs"),
489        }
490    }
491
492    #[test]
493    fn parse_refresh_interval_uses_default_and_rejects_invalid_values() {
494        assert_eq!(
495            parse_refresh_interval(None).expect("default interval"),
496            Duration::from_secs(30)
497        );
498        assert!(parse_refresh_interval(Some("not-a-duration".into())).is_err());
499    }
500
501    #[test]
502    fn ensure_paths_exist_creates_expected_directories() {
503        let temp = TempDir::new().expect("tempdir");
504        let paths = paths(&temp);
505        ensure_paths_exist(&paths).expect("create directories");
506
507        assert!(paths.greentic_root.is_dir());
508        assert!(paths.state_dir.is_dir());
509        assert!(paths.cache_dir.is_dir());
510        assert!(paths.logs_dir.is_dir());
511    }
512
513    #[test]
514    fn maybe_write_gtbind_index_writes_locator_index() {
515        let temp = TempDir::new().expect("tempdir");
516        let paths = paths(&temp);
517        fs::create_dir_all(&paths.greentic_root).expect("greentic root");
518        let mut pack = PackConfig::default_for_paths(&paths).expect("pack config");
519        let tenant_bindings = HashMap::from([(
520            "demo".to_string(),
521            TenantBindings {
522                tenant: "demo".into(),
523                packs: vec![
524                    PackBinding {
525                        pack_id: "pack.main".into(),
526                        pack_ref: "pack.main@1.0.0".into(),
527                        pack_locator: Some("fs:///packs/main.gtpack".into()),
528                        flows: vec!["main".into()],
529                    },
530                    PackBinding {
531                        pack_id: "pack.overlay".into(),
532                        pack_ref: "pack.overlay@sha256:abcd".into(),
533                        pack_locator: Some("fs:///packs/overlay.gtpack".into()),
534                        flows: vec![],
535                    },
536                ],
537                env_passthrough: vec![],
538            },
539        )]);
540
541        maybe_write_gtbind_index(&tenant_bindings, &paths, &mut pack).expect("write gtbind index");
542
543        let index_path = paths.greentic_root.join("packs").join("gtbind.index.json");
544        let json: serde_json::Value =
545            serde_json::from_slice(&fs::read(&index_path).expect("read index")).expect("json");
546        assert_eq!(
547            json["demo"]["main_pack"]["locator"],
548            "fs:///packs/main.gtpack"
549        );
550        assert_eq!(json["demo"]["overlays"][0]["digest"], "sha256:abcd");
551        match pack.index_location {
552            runner_core::env::IndexLocation::File(path) => assert_eq!(path, index_path),
553            runner_core::env::IndexLocation::Remote(_) => panic!("expected generated file index"),
554        }
555    }
556
557    #[test]
558    fn pack_config_from_prefers_configured_pack_source() {
559        let temp = TempDir::new().expect("tempdir");
560        let paths = paths(&temp);
561        let packs = Some(PacksConfig {
562            source: PackSourceConfig::HttpIndex {
563                url: "https://example.com/index.json".into(),
564            },
565            cache_dir: temp.path().join("packs-cache"),
566            index_cache_ttl_secs: None,
567            trust: Some(PackTrustConfig {
568                public_keys: vec!["ed25519:test".into()],
569                require_signatures: true,
570            }),
571        });
572        let config = pack_config_from(&packs, &paths, &NetworkConfig::default())
573            .expect("pack config from packs");
574
575        match config.index_location {
576            runner_core::env::IndexLocation::Remote(url) => {
577                assert_eq!(url.as_str(), "https://example.com/index.json");
578            }
579            runner_core::env::IndexLocation::File(_) => panic!("expected remote index"),
580        }
581        assert_eq!(config.public_key.as_deref(), Some("ed25519:test"));
582        assert!(config.network.is_some());
583    }
584
585    #[test]
586    fn default_wasi_policy_sets_expected_env_and_preopens() {
587        let temp = TempDir::new().expect("tempdir");
588        let paths = paths(&temp);
589        let policy = default_wasi_policy(&paths);
590
591        assert_eq!(
592            policy.env_set.get("GREENTIC_ROOT"),
593            Some(&paths.greentic_root.display().to_string())
594        );
595        assert_eq!(policy.preopens.len(), 3);
596        assert_eq!(policy.preopens[0].guest_path, "/state");
597        assert_eq!(policy.preopens[1].guest_path, "/cache");
598        assert_eq!(policy.preopens[2].guest_path, "/logs");
599    }
600
601    #[test]
602    fn telemetry_from_is_disabled_without_exporter() {
603        assert!(telemetry_from(&TelemetryConfig::default()).is_none());
604    }
605}
greentic_runner_host/lib.rs

greentic_runner_host/
lib.rs
