lenso_module_management/
service_installation.rs

1use chrono::{DateTime, Utc};
2use fs2::FileExt as _;
3use lenso_contracts::{ServiceResponsibilityProfile, digest_json};
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6use std::collections::{BTreeMap, BTreeSet};
7use std::fs::{self, OpenOptions};
8use std::io::Write as _;
9use std::path::{Path, PathBuf};
10use thiserror::Error;
11
12pub const SERVICE_INSTALLATION_SET_PROTOCOL: &str = "lenso.service-installations.v1";
13pub const SERVICE_INSTALLATION_PLAN_PROTOCOL: &str = "lenso.service-install-plan.v1";
14pub const SERVICE_INSTALLATION_RECEIPT_PROTOCOL: &str = "lenso.service-install-receipt.v1";
15
16#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema)]
17#[serde(deny_unknown_fields)]
18pub struct ServiceReference {
19    pub system_id: String,
20    pub service_id: String,
21}
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
24#[serde(rename_all = "snake_case")]
25pub enum ServiceDesiredMode {
26    Active,
27    Inactive,
28}
29
30#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
31#[serde(deny_unknown_fields)]
32pub struct InstalledServiceRelease {
33    pub version: String,
34    pub digest: String,
35    pub immutable_locator: String,
36}
37
38#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
39#[serde(deny_unknown_fields)]
40pub struct InstalledServiceExport {
41    pub export_key: String,
42    pub module_id: String,
43    pub module_version: String,
44    pub module_release_digest: String,
45    pub manifest_digest: String,
46    #[serde(default, skip_serializing_if = "Vec::is_empty")]
47    pub contract_digests: Vec<String>,
48}
49
50#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
51#[serde(rename_all = "snake_case")]
52pub enum ConfigActivationIntent {
53    Prepare,
54    Activate,
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
58#[serde(deny_unknown_fields)]
59pub struct ServiceConfigBinding {
60    pub owner_id: String,
61    pub config_contract_digest: String,
62    pub config_revision_id: String,
63    pub config_revision_digest: String,
64    pub activation: ConfigActivationIntent,
65    #[serde(default, skip_serializing_if = "Vec::is_empty")]
66    pub secret_references: Vec<String>,
67}
68
69#[derive(
70    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
71)]
72#[serde(rename_all = "snake_case")]
73pub enum ServiceTransportBinding {
74    ProviderHttpJson,
75    ProviderGrpc,
76    DirectHttp,
77    DirectGrpc,
78    Event,
79    SystemPlane,
80}
81
82#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
83#[serde(deny_unknown_fields)]
84pub struct StaticEndpointDeclaration {
85    pub address: String,
86    pub binding: ServiceTransportBinding,
87    #[serde(default, skip_serializing_if = "Option::is_none")]
88    pub region: Option<String>,
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub failure_domain: Option<String>,
91    #[serde(default)]
92    pub priority: u32,
93    #[serde(default = "default_weight")]
94    pub weight: u32,
95}
96
97const fn default_weight() -> u32 {
98    1
99}
100
101#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
102#[serde(tag = "kind", rename_all = "snake_case")]
103pub enum EndpointResolverSource {
104    Static {
105        endpoints: Vec<StaticEndpointDeclaration>,
106    },
107    LocalProcess {
108        source_id: String,
109    },
110    Adapter {
111        adapter_id: String,
112        #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
113        public_config: BTreeMap<String, String>,
114        #[serde(default, skip_serializing_if = "Vec::is_empty")]
115        secret_references: Vec<String>,
116    },
117}
118
119#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
120#[serde(deny_unknown_fields)]
121pub struct ServiceIdentityPolicy {
122    pub principal: String,
123    pub audience: String,
124    pub trust_profile: String,
125    #[serde(default, skip_serializing_if = "Vec::is_empty")]
126    pub credential_references: Vec<String>,
127}
128
129#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
130#[serde(deny_unknown_fields)]
131pub struct EndpointSelectionPolicy {
132    #[serde(default, skip_serializing_if = "Vec::is_empty")]
133    pub preferred_regions: Vec<String>,
134    #[serde(default)]
135    pub require_distinct_failure_domains: bool,
136}
137
138#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
139#[serde(deny_unknown_fields)]
140pub struct EndpointCachePolicy {
141    pub maximum_age_seconds: u64,
142    #[serde(default, skip_serializing_if = "Option::is_none")]
143    pub stale_if_source_unavailable_seconds: Option<u64>,
144}
145
146#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
147#[serde(deny_unknown_fields)]
148pub struct EndpointBinding {
149    pub binding_id: String,
150    pub service_ref: ServiceReference,
151    pub resolver_source: EndpointResolverSource,
152    pub allowed_bindings: Vec<ServiceTransportBinding>,
153    pub identity_policy: ServiceIdentityPolicy,
154    #[serde(default)]
155    pub selection_policy: EndpointSelectionPolicy,
156    pub cache_policy: EndpointCachePolicy,
157}
158
159#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
160#[serde(tag = "kind", rename_all = "snake_case")]
161pub enum ServiceLifecycleBinding {
162    External {
163        deployment_reference: String,
164        observation_adapter_id: String,
165        #[serde(default, skip_serializing_if = "Option::is_none")]
166        operation_adapter_id: Option<String>,
167    },
168    LocalSupervisor {
169        supervisor_id: String,
170        workload_artifact_digests: Vec<String>,
171        working_directory: String,
172        readiness_timeout_seconds: u64,
173        shutdown_timeout_seconds: u64,
174    },
175}
176
177#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
178#[serde(deny_unknown_fields)]
179pub struct ServiceInstallation {
180    pub service_ref: ServiceReference,
181    pub profile: ServiceResponsibilityProfile,
182    pub desired_mode: ServiceDesiredMode,
183    pub service_release: InstalledServiceRelease,
184    pub exports: Vec<InstalledServiceExport>,
185    #[serde(default, skip_serializing_if = "Vec::is_empty")]
186    pub config_bindings: Vec<ServiceConfigBinding>,
187    pub endpoint_binding: EndpointBinding,
188    pub lifecycle_binding: ServiceLifecycleBinding,
189}
190
191#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
192#[serde(deny_unknown_fields)]
193pub struct ServiceInstallationSet {
194    pub protocol: String,
195    pub system_id: String,
196    pub environment_id: String,
197    pub revision: u64,
198    #[serde(default, skip_serializing_if = "Option::is_none")]
199    pub previous_state_digest: Option<String>,
200    pub services: Vec<ServiceInstallation>,
201}
202
203impl ServiceInstallationSet {
204    #[must_use]
205    pub fn empty(system_id: impl Into<String>, environment_id: impl Into<String>) -> Self {
206        Self {
207            protocol: SERVICE_INSTALLATION_SET_PROTOCOL.to_owned(),
208            system_id: system_id.into(),
209            environment_id: environment_id.into(),
210            revision: 0,
211            previous_state_digest: None,
212            services: Vec::new(),
213        }
214    }
215}
216
217#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
218#[serde(tag = "kind", rename_all = "snake_case")]
219pub enum ServiceInstallationChange {
220    Install { installation: ServiceInstallation },
221    Uninstall { service_ref: ServiceReference },
222}
223
224#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
225#[serde(rename_all = "snake_case")]
226pub enum ServiceInstallationPlanKind {
227    Install,
228    Update,
229    Reuse,
230    Uninstall,
231}
232
233#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
234#[serde(deny_unknown_fields)]
235pub struct ServiceInstallationPlan {
236    pub protocol: String,
237    pub plan_id: String,
238    pub plan_digest: String,
239    pub system_id: String,
240    pub environment_id: String,
241    pub kind: ServiceInstallationPlanKind,
242    pub change: ServiceInstallationChange,
243    pub expected_revision: u64,
244    pub expected_state_digest: String,
245    pub target_revision: u64,
246    pub target_state_digest: String,
247    pub target: ServiceInstallationSet,
248    pub required_authority: String,
249    pub next_actions: Vec<String>,
250    pub created_at: DateTime<Utc>,
251}
252
253#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
254#[serde(rename_all = "snake_case")]
255pub enum ServiceInstallationOutcome {
256    AppliedNeedsAttention,
257    Reused,
258    Removed,
259}
260
261#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
262#[serde(deny_unknown_fields)]
263pub struct ServiceInstallationReceipt {
264    pub protocol: String,
265    pub receipt_id: String,
266    pub operation_id: String,
267    pub plan_id: String,
268    pub plan_digest: String,
269    pub actor_id: String,
270    pub verified_authorities: Vec<String>,
271    pub system_id: String,
272    pub environment_id: String,
273    pub service_ref: ServiceReference,
274    pub prior_revision: u64,
275    pub target_revision: u64,
276    pub prior_state_digest: String,
277    pub target_state_digest: String,
278    pub outcome: ServiceInstallationOutcome,
279    pub reasons: Vec<String>,
280    pub next_actions: Vec<String>,
281    pub committed_at: DateTime<Utc>,
282}
283
284#[derive(Debug, Error)]
285pub enum ServiceInstallationError {
286    #[error("Service Installation contract is invalid: {0}")]
287    InvalidContract(String),
288    #[error("Service Installation state is stale")]
289    StaleState,
290    #[error("Service Installation operation requires authority `{0}`")]
291    MissingAuthority(String),
292    #[error("Service Installation operation identity is unsafe")]
293    UnsafeOperationIdentity,
294    #[error("Service Installation I/O failed: {0}")]
295    Io(#[from] std::io::Error),
296    #[error("Service Installation JSON failed: {0}")]
297    Json(#[from] serde_json::Error),
298}
299
300#[derive(Debug, Clone)]
301pub struct WorkspaceServiceInstallationManager {
302    root: PathBuf,
303    system_id: String,
304    environment_id: String,
305}
306
307impl WorkspaceServiceInstallationManager {
308    pub fn new(
309        root: impl Into<PathBuf>,
310        system_id: impl Into<String>,
311        environment_id: impl Into<String>,
312    ) -> Self {
313        Self {
314            root: root.into(),
315            system_id: system_id.into(),
316            environment_id: environment_id.into(),
317        }
318    }
319
320    pub fn snapshot(&self) -> Result<ServiceInstallationSet, ServiceInstallationError> {
321        let path = self.state_path()?;
322        match fs::read(path) {
323            Ok(bytes) => {
324                let state: ServiceInstallationSet = serde_json::from_slice(&bytes)?;
325                validate_service_installation_set(&state)?;
326                if state.system_id != self.system_id || state.environment_id != self.environment_id
327                {
328                    return invalid("Service Installation Set scope differs from target scope");
329                }
330                Ok(state)
331            }
332            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(
333                ServiceInstallationSet::empty(&self.system_id, &self.environment_id),
334            ),
335            Err(error) => Err(error.into()),
336        }
337    }
338
339    pub fn preview(
340        &self,
341        change: ServiceInstallationChange,
342        created_at: DateTime<Utc>,
343    ) -> Result<ServiceInstallationPlan, ServiceInstallationError> {
344        let current = self.snapshot()?;
345        plan_service_installation(&current, change, created_at)
346    }
347
348    pub fn apply(
349        &self,
350        operation_id: &str,
351        plan: &ServiceInstallationPlan,
352        actor_id: &str,
353        authorities: &BTreeSet<String>,
354        now: DateTime<Utc>,
355    ) -> Result<ServiceInstallationReceipt, ServiceInstallationError> {
356        if !safe_id(operation_id) {
357            return Err(ServiceInstallationError::UnsafeOperationIdentity);
358        }
359        if !authorities.contains(&plan.required_authority) {
360            return Err(ServiceInstallationError::MissingAuthority(
361                plan.required_authority.clone(),
362            ));
363        }
364        validate_service_installation_plan(plan)?;
365        let receipt_path = self.receipt_path(operation_id)?;
366        if let Ok(bytes) = fs::read(&receipt_path) {
367            let receipt: ServiceInstallationReceipt = serde_json::from_slice(&bytes)?;
368            if receipt.protocol != SERVICE_INSTALLATION_RECEIPT_PROTOCOL
369                || receipt.operation_id != operation_id
370                || receipt.plan_id != plan.plan_id
371                || receipt.plan_digest != plan.plan_digest
372                || receipt.target_state_digest != plan.target_state_digest
373            {
374                return Err(ServiceInstallationError::StaleState);
375            }
376            return Ok(receipt);
377        }
378
379        let lock_path = self.environment_root()?.join("service-installations.lock");
380        fs::create_dir_all(self.environment_root()?)?;
381        let lock = OpenOptions::new()
382            .create(true)
383            .read(true)
384            .write(true)
385            .open(lock_path)?;
386        lock.lock_exclusive()?;
387        let result = self.apply_locked(operation_id, plan, actor_id, authorities, now);
388        let unlock_result = lock.unlock();
389        result.and_then(|receipt| {
390            unlock_result?;
391            Ok(receipt)
392        })
393    }
394
395    fn apply_locked(
396        &self,
397        operation_id: &str,
398        plan: &ServiceInstallationPlan,
399        actor_id: &str,
400        authorities: &BTreeSet<String>,
401        now: DateTime<Utc>,
402    ) -> Result<ServiceInstallationReceipt, ServiceInstallationError> {
403        let current = self.snapshot()?;
404        let current_digest = service_installation_set_digest(&current)?;
405        if current.revision == plan.target_revision && current_digest == plan.target_state_digest {
406            let receipt =
407                service_installation_receipt(operation_id, plan, actor_id, authorities, now)?;
408            atomic_write_json(&self.receipt_path(operation_id)?, &receipt)?;
409            return Ok(receipt);
410        }
411        if current.revision != plan.expected_revision
412            || current_digest != plan.expected_state_digest
413        {
414            return Err(ServiceInstallationError::StaleState);
415        }
416        let verified = plan_service_installation(&current, plan.change.clone(), plan.created_at)?;
417        if &verified != plan {
418            return Err(ServiceInstallationError::StaleState);
419        }
420        if plan.kind != ServiceInstallationPlanKind::Reuse {
421            atomic_write_json(&self.state_path()?, &plan.target)?;
422        }
423        let receipt = service_installation_receipt(operation_id, plan, actor_id, authorities, now)?;
424        atomic_write_json(&self.receipt_path(operation_id)?, &receipt)?;
425        Ok(receipt)
426    }
427
428    fn environment_root(&self) -> Result<PathBuf, ServiceInstallationError> {
429        if !safe_segment(&self.environment_id) {
430            return invalid("environment identity is unsafe");
431        }
432        Ok(self
433            .root
434            .join(".lenso/environments")
435            .join(&self.environment_id))
436    }
437
438    fn state_path(&self) -> Result<PathBuf, ServiceInstallationError> {
439        Ok(self.environment_root()?.join("service-installations.json"))
440    }
441
442    fn receipt_path(&self, operation_id: &str) -> Result<PathBuf, ServiceInstallationError> {
443        if !safe_id(operation_id) {
444            return Err(ServiceInstallationError::UnsafeOperationIdentity);
445        }
446        Ok(self
447            .environment_root()?
448            .join("service-install-receipts")
449            .join(format!("{operation_id}.json")))
450    }
451}
452
453pub fn plan_service_installation(
454    current: &ServiceInstallationSet,
455    change: ServiceInstallationChange,
456    created_at: DateTime<Utc>,
457) -> Result<ServiceInstallationPlan, ServiceInstallationError> {
458    validate_service_installation_set(current)?;
459    let expected_state_digest = service_installation_set_digest(current)?;
460    let mut target = current.clone();
461    let kind = match &change {
462        ServiceInstallationChange::Install { installation } => {
463            validate_service_installation(installation)?;
464            if installation.service_ref.system_id != current.system_id {
465                return invalid("Service Reference system differs from Installation Set");
466            }
467            match target
468                .services
469                .binary_search_by(|candidate| candidate.service_ref.cmp(&installation.service_ref))
470            {
471                Ok(index) if target.services[index] == *installation => {
472                    ServiceInstallationPlanKind::Reuse
473                }
474                Ok(index) => {
475                    if target.services[index].profile != installation.profile {
476                        return invalid(
477                            "Service responsibility profile replacement requires a separate plan",
478                        );
479                    }
480                    target.services[index] = installation.clone();
481                    ServiceInstallationPlanKind::Update
482                }
483                Err(index) => {
484                    target.services.insert(index, installation.clone());
485                    ServiceInstallationPlanKind::Install
486                }
487            }
488        }
489        ServiceInstallationChange::Uninstall { service_ref } => {
490            if service_ref.system_id != current.system_id {
491                return invalid("Service Reference system differs from Installation Set");
492            }
493            let index = target
494                .services
495                .binary_search_by(|candidate| candidate.service_ref.cmp(service_ref))
496                .map_err(|_| {
497                    ServiceInstallationError::InvalidContract("Service is not installed".to_owned())
498                })?;
499            target.services.remove(index);
500            ServiceInstallationPlanKind::Uninstall
501        }
502    };
503    if kind != ServiceInstallationPlanKind::Reuse {
504        target.previous_state_digest = Some(expected_state_digest.clone());
505        target.revision = current.revision.saturating_add(1);
506    }
507    let target_state_digest = service_installation_set_digest(&target)?;
508    let identity = digest_json(&(
509        current.system_id.as_str(),
510        current.environment_id.as_str(),
511        &change,
512        current.revision,
513        expected_state_digest.as_str(),
514        target_state_digest.as_str(),
515    ))?;
516    let mut plan = ServiceInstallationPlan {
517        protocol: SERVICE_INSTALLATION_PLAN_PROTOCOL.to_owned(),
518        plan_id: format!("service-install-plan:{}", &identity[7..23]),
519        plan_digest: String::new(),
520        system_id: current.system_id.clone(),
521        environment_id: current.environment_id.clone(),
522        kind,
523        change,
524        expected_revision: current.revision,
525        expected_state_digest,
526        target_revision: target.revision,
527        target_state_digest,
528        target,
529        required_authority: "service.manage".to_owned(),
530        next_actions: vec![
531            "review_service_installation_plan".to_owned(),
532            "apply_service_installation_plan".to_owned(),
533        ],
534        created_at,
535    };
536    plan.plan_digest = service_installation_plan_digest(&plan)?;
537    validate_service_installation_plan(&plan)?;
538    Ok(plan)
539}
540
541pub fn service_installation_set_digest(
542    state: &ServiceInstallationSet,
543) -> Result<String, serde_json::Error> {
544    digest_json(state)
545}
546
547pub fn service_installation_plan_digest(
548    plan: &ServiceInstallationPlan,
549) -> Result<String, serde_json::Error> {
550    let mut content = plan.clone();
551    content.plan_digest.clear();
552    digest_json(&content)
553}
554
555pub fn validate_service_installation_plan(
556    plan: &ServiceInstallationPlan,
557) -> Result<(), ServiceInstallationError> {
558    if plan.protocol != SERVICE_INSTALLATION_PLAN_PROTOCOL
559        || service_installation_plan_digest(plan)? != plan.plan_digest
560        || plan.target_state_digest != service_installation_set_digest(&plan.target)?
561        || plan.target.system_id != plan.system_id
562        || plan.target.environment_id != plan.environment_id
563        || plan.required_authority != "service.manage"
564    {
565        return invalid("Service Installation Plan identity or digest is invalid");
566    }
567    validate_service_installation_set(&plan.target)
568}
569
570pub fn validate_service_installation_set(
571    state: &ServiceInstallationSet,
572) -> Result<(), ServiceInstallationError> {
573    if state.protocol != SERVICE_INSTALLATION_SET_PROTOCOL
574        || !safe_identity(&state.system_id)
575        || !safe_segment(&state.environment_id)
576        || state
577            .previous_state_digest
578            .as_deref()
579            .is_some_and(|digest| !valid_digest(digest))
580    {
581        return invalid("Service Installation Set identity, protocol, or digest is invalid");
582    }
583    require_sorted_unique(
584        state.services.iter().map(|service| &service.service_ref),
585        "Service References",
586    )?;
587    for service in &state.services {
588        validate_service_installation(service)?;
589        if service.service_ref.system_id != state.system_id {
590            return invalid("installed Service belongs to another System");
591        }
592    }
593    Ok(())
594}
595
596pub fn validate_service_installation(
597    installation: &ServiceInstallation,
598) -> Result<(), ServiceInstallationError> {
599    if !safe_identity(&installation.service_ref.system_id)
600        || !safe_identity(&installation.service_ref.service_id)
601        || semver::Version::parse(&installation.service_release.version).is_err()
602        || !valid_digest(&installation.service_release.digest)
603        || installation
604            .service_release
605            .immutable_locator
606            .trim()
607            .is_empty()
608        || installation.endpoint_binding.service_ref != installation.service_ref
609        || installation.exports.is_empty()
610        || installation.endpoint_binding.binding_id.trim().is_empty()
611        || installation.endpoint_binding.allowed_bindings.is_empty()
612        || installation
613            .endpoint_binding
614            .cache_policy
615            .maximum_age_seconds
616            == 0
617        || installation
618            .endpoint_binding
619            .identity_policy
620            .principal
621            .trim()
622            .is_empty()
623        || installation
624            .endpoint_binding
625            .identity_policy
626            .audience
627            .trim()
628            .is_empty()
629        || installation
630            .endpoint_binding
631            .identity_policy
632            .trust_profile
633            .trim()
634            .is_empty()
635    {
636        return invalid(
637            "installed Service identity, release, endpoint, or identity policy is invalid",
638        );
639    }
640    require_sorted_unique(
641        installation
642            .exports
643            .iter()
644            .map(|export| export.export_key.as_str()),
645        "Service export keys",
646    )?;
647    require_sorted_unique(
648        installation.endpoint_binding.allowed_bindings.iter(),
649        "allowed endpoint bindings",
650    )?;
651    for export in &installation.exports {
652        if export.export_key.trim().is_empty()
653            || !valid_module_id(&export.module_id)
654            || semver::Version::parse(&export.module_version).is_err()
655            || !valid_digest(&export.module_release_digest)
656            || !valid_digest(&export.manifest_digest)
657            || export
658                .contract_digests
659                .iter()
660                .any(|digest| !valid_digest(digest))
661        {
662            return invalid("installed Service export is invalid");
663        }
664        require_sorted_unique(
665            export.contract_digests.iter(),
666            "Service export contract digests",
667        )?;
668    }
669    for binding in &installation.config_bindings {
670        if binding.owner_id.trim().is_empty()
671            || !valid_digest(&binding.config_contract_digest)
672            || binding.config_revision_id.trim().is_empty()
673            || !valid_digest(&binding.config_revision_digest)
674        {
675            return invalid("Service Config binding is invalid");
676        }
677        require_sorted_unique(binding.secret_references.iter(), "Secret References")?;
678    }
679    match &installation.endpoint_binding.resolver_source {
680        EndpointResolverSource::Static { endpoints } => {
681            if endpoints.is_empty()
682                || endpoints.iter().any(|endpoint| {
683                    endpoint.address.trim().is_empty()
684                        || endpoint.weight == 0
685                        || !installation
686                            .endpoint_binding
687                            .allowed_bindings
688                            .contains(&endpoint.binding)
689                })
690            {
691                return invalid("static Endpoint source is invalid");
692            }
693        }
694        EndpointResolverSource::LocalProcess { source_id } => {
695            if source_id.trim().is_empty() {
696                return invalid("local process Endpoint source is invalid");
697            }
698        }
699        EndpointResolverSource::Adapter {
700            adapter_id,
701            public_config,
702            secret_references,
703        } => {
704            if adapter_id.trim().is_empty()
705                || public_config.iter().any(|(key, value)| {
706                    key.trim().is_empty()
707                        || value.trim().is_empty()
708                        || secret_shaped(key)
709                        || secret_shaped(value)
710                })
711            {
712                return invalid("external Endpoint adapter configuration is invalid");
713            }
714            require_sorted_unique(secret_references.iter(), "Endpoint Secret References")?;
715        }
716    }
717    Ok(())
718}
719
720fn service_installation_receipt(
721    operation_id: &str,
722    plan: &ServiceInstallationPlan,
723    actor_id: &str,
724    authorities: &BTreeSet<String>,
725    now: DateTime<Utc>,
726) -> Result<ServiceInstallationReceipt, ServiceInstallationError> {
727    let service_ref = match &plan.change {
728        ServiceInstallationChange::Install { installation } => installation.service_ref.clone(),
729        ServiceInstallationChange::Uninstall { service_ref } => service_ref.clone(),
730    };
731    let (outcome, reasons, next_actions) = match plan.kind {
732        ServiceInstallationPlanKind::Reuse => (
733            ServiceInstallationOutcome::Reused,
734            vec!["desired_service_installation_already_matches".to_owned()],
735            Vec::new(),
736        ),
737        ServiceInstallationPlanKind::Uninstall => (
738            ServiceInstallationOutcome::Removed,
739            vec!["desired_service_installation_removed".to_owned()],
740            vec!["preserve_deployment_data_and_observations".to_owned()],
741        ),
742        ServiceInstallationPlanKind::Install | ServiceInstallationPlanKind::Update => (
743            ServiceInstallationOutcome::AppliedNeedsAttention,
744            vec!["runtime_readiness_requires_fresh_observation".to_owned()],
745            vec!["observe_service_identity_and_readiness".to_owned()],
746        ),
747    };
748    let mut verified_authorities = authorities.iter().cloned().collect::<Vec<_>>();
749    verified_authorities.sort();
750    let receipt_seed = digest_json(&(operation_id, plan.plan_digest.as_str(), actor_id, now))?;
751    Ok(ServiceInstallationReceipt {
752        protocol: SERVICE_INSTALLATION_RECEIPT_PROTOCOL.to_owned(),
753        receipt_id: format!("service-install-receipt:{}", &receipt_seed[7..23]),
754        operation_id: operation_id.to_owned(),
755        plan_id: plan.plan_id.clone(),
756        plan_digest: plan.plan_digest.clone(),
757        actor_id: actor_id.to_owned(),
758        verified_authorities,
759        system_id: plan.system_id.clone(),
760        environment_id: plan.environment_id.clone(),
761        service_ref,
762        prior_revision: plan.expected_revision,
763        target_revision: plan.target_revision,
764        prior_state_digest: plan.expected_state_digest.clone(),
765        target_state_digest: plan.target_state_digest.clone(),
766        outcome,
767        reasons,
768        next_actions,
769        committed_at: now,
770    })
771}
772
773fn atomic_write_json<T: Serialize>(path: &Path, value: &T) -> Result<(), ServiceInstallationError> {
774    let parent = path.parent().ok_or_else(|| {
775        ServiceInstallationError::InvalidContract("state path has no parent".to_owned())
776    })?;
777    fs::create_dir_all(parent)?;
778    let temporary = path.with_extension(format!("json.tmp-{}", std::process::id()));
779    let bytes = serde_json::to_vec_pretty(value)?;
780    let mut file = OpenOptions::new()
781        .create(true)
782        .truncate(true)
783        .write(true)
784        .open(&temporary)?;
785    file.write_all(&bytes)?;
786    file.sync_all()?;
787    fs::rename(&temporary, path)?;
788    OpenOptions::new().read(true).open(parent)?.sync_all()?;
789    Ok(())
790}
791
792fn require_sorted_unique<'a, T: Ord + ?Sized + 'a>(
793    values: impl IntoIterator<Item = &'a T>,
794    field: &str,
795) -> Result<(), ServiceInstallationError> {
796    let values = values.into_iter().collect::<Vec<_>>();
797    if values.windows(2).any(|pair| pair[0] >= pair[1]) {
798        return invalid(format!("{field} must be sorted and unique"));
799    }
800    Ok(())
801}
802
803fn valid_digest(value: &str) -> bool {
804    value.strip_prefix("sha256:").is_some_and(|hex| {
805        hex.len() == 64
806            && hex
807                .bytes()
808                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
809    })
810}
811
812fn safe_identity(value: &str) -> bool {
813    !value.is_empty()
814        && value.len() <= 255
815        && value.chars().all(|character| {
816            character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '.' | '/')
817        })
818}
819
820fn safe_segment(value: &str) -> bool {
821    safe_identity(value) && !value.contains('/') && value != "." && value != ".."
822}
823
824fn safe_id(value: &str) -> bool {
825    safe_segment(value) && value.len() <= 180
826}
827
828fn valid_module_id(value: &str) -> bool {
829    value.split_once('/').is_some_and(|(namespace, name)| {
830        !namespace.is_empty()
831            && !name.is_empty()
832            && !name.contains('/')
833            && safe_identity(namespace)
834            && safe_identity(name)
835    })
836}
837
838fn secret_shaped(value: &str) -> bool {
839    let normalized = value.to_ascii_lowercase();
840    ["secret", "password", "credential", "privatekey", "token"]
841        .iter()
842        .any(|needle| normalized.contains(needle))
843}
844
845fn invalid<T>(message: impl Into<String>) -> Result<T, ServiceInstallationError> {
846    Err(ServiceInstallationError::InvalidContract(message.into()))
847}
lenso_module_management/service_installation.rs

lenso_module_management/
service_installation.rs
