platform_system_plane/
lib.rs

1//! Capability-neutral System Plane Core routing, registration, and negotiation.
2
3mod enrollment;
4mod module_operations;
5
6pub use enrollment::*;
7pub use module_operations::*;
8
9use axum::{
10    Extension, Json,
11    extract::FromRequestParts,
12    http::{HeaderMap, HeaderValue, StatusCode, header},
13    response::{IntoResponse, Response},
14};
15use lenso_service::{
16    AuthenticatedTransportBinding, WorkloadIdentityProvider, WorkloadIdentityVerification,
17    system_plane::{
18        CORE_PROTOCOL, CapabilityAdvertisement, CoreDocument, CoreIssue, validate_core_document,
19    },
20};
21use serde::Serialize;
22use std::{
23    collections::{BTreeSet, HashSet},
24    fmt,
25    sync::Arc,
26    time::{SystemTime, UNIX_EPOCH},
27};
28use utoipa::ToSchema;
29use utoipa_axum::{router::OpenApiRouter, routes};
30
31#[derive(Debug, Clone)]
32pub struct SystemPlaneRegistry {
33    document: Arc<CoreDocument>,
34}
35
36impl SystemPlaneRegistry {
37    pub fn new(document: CoreDocument) -> Result<Self, Vec<CoreIssue>> {
38        let issues = validate_core_document(&document);
39        if issues.is_empty() {
40            Ok(Self {
41                document: Arc::new(document),
42            })
43        } else {
44            Err(issues)
45        }
46    }
47
48    #[must_use]
49    pub fn document(&self) -> &CoreDocument {
50        &self.document
51    }
52
53    #[must_use]
54    pub fn negotiate(&self, requirements: &[CapabilityRequirement]) -> CapabilityNegotiation {
55        negotiate_capabilities(self.document(), requirements)
56    }
57}
58
59#[derive(Debug, Clone)]
60pub struct SystemPlaneRegistryBuilder {
61    document: CoreDocument,
62}
63
64impl SystemPlaneRegistryBuilder {
65    #[must_use]
66    pub fn new(
67        service_id: impl Into<String>,
68        service_principal: impl Into<String>,
69        service_revision: impl Into<String>,
70    ) -> Self {
71        Self {
72            document: CoreDocument {
73                protocol: CORE_PROTOCOL.to_owned(),
74                service_id: service_id.into(),
75                service_principal: service_principal.into(),
76                service_revision: service_revision.into(),
77                capabilities: Vec::new(),
78            },
79        }
80    }
81
82    #[must_use]
83    pub fn register(mut self, capability: CapabilityAdvertisement) -> Self {
84        self.document.capabilities.push(capability);
85        self
86    }
87
88    pub fn build(mut self) -> Result<SystemPlaneRegistry, Vec<CoreIssue>> {
89        self.document
90            .capabilities
91            .sort_by(|left, right| left.contract_id.cmp(&right.contract_id));
92        SystemPlaneRegistry::new(self.document)
93    }
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
97pub struct CapabilityRequirement {
98    pub capability_id: String,
99    pub supported_major_versions: BTreeSet<u32>,
100    pub required_feature_ids: BTreeSet<String>,
101    pub accepted_schema_digests: BTreeSet<String>,
102}
103
104impl CapabilityRequirement {
105    #[must_use]
106    pub fn new(
107        capability_id: impl Into<String>,
108        supported_major_versions: impl IntoIterator<Item = u32>,
109    ) -> Self {
110        Self {
111            capability_id: capability_id.into(),
112            supported_major_versions: supported_major_versions.into_iter().collect(),
113            required_feature_ids: BTreeSet::new(),
114            accepted_schema_digests: BTreeSet::new(),
115        }
116    }
117
118    #[must_use]
119    pub fn requiring_features(
120        mut self,
121        feature_ids: impl IntoIterator<Item = impl Into<String>>,
122    ) -> Self {
123        self.required_feature_ids = feature_ids.into_iter().map(Into::into).collect();
124        self
125    }
126
127    #[must_use]
128    pub fn accepting_schema_digests(
129        mut self,
130        digests: impl IntoIterator<Item = impl Into<String>>,
131    ) -> Self {
132        self.accepted_schema_digests = digests.into_iter().map(Into::into).collect();
133        self
134    }
135}
136
137#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
138#[serde(rename_all = "snake_case")]
139pub enum CapabilityNegotiationIssueCode {
140    InvalidRequirement,
141    DuplicateRequirement,
142    MissingCapability,
143    UnsupportedMajorVersion,
144    MissingRequiredFeature,
145    SchemaDigestMismatch,
146}
147
148#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
149#[serde(rename_all = "camelCase")]
150pub struct CapabilityNegotiationIssue {
151    pub code: CapabilityNegotiationIssueCode,
152    pub capability_id: String,
153    pub message: String,
154}
155
156#[derive(Debug, Clone, PartialEq, Eq)]
157pub struct NegotiatedCapability {
158    pub capability_id: String,
159    pub advertisement: CapabilityAdvertisement,
160}
161
162#[derive(Debug, Clone, PartialEq, Eq)]
163pub struct CapabilityNegotiation {
164    pub accepted: Vec<NegotiatedCapability>,
165    pub issues: Vec<CapabilityNegotiationIssue>,
166}
167
168impl CapabilityNegotiation {
169    #[must_use]
170    pub const fn is_compatible(&self) -> bool {
171        self.issues.is_empty()
172    }
173}
174
175#[must_use]
176pub fn negotiate_capabilities(
177    document: &CoreDocument,
178    requirements: &[CapabilityRequirement],
179) -> CapabilityNegotiation {
180    let mut accepted = Vec::new();
181    let mut issues = Vec::new();
182    let mut seen = HashSet::new();
183
184    for requirement in requirements {
185        if !valid_capability_id(&requirement.capability_id)
186            || requirement.supported_major_versions.is_empty()
187            || requirement.supported_major_versions.contains(&0)
188        {
189            negotiation_issue(
190                &mut issues,
191                CapabilityNegotiationIssueCode::InvalidRequirement,
192                requirement,
193                "Capability requirements need a canonical identifier and at least one positive major version",
194            );
195            continue;
196        }
197        if !seen.insert(requirement.capability_id.as_str()) {
198            negotiation_issue(
199                &mut issues,
200                CapabilityNegotiationIssueCode::DuplicateRequirement,
201                requirement,
202                "Each capability may be negotiated once",
203            );
204            continue;
205        }
206
207        let prefix = format!("lenso.system-plane.{}.v", requirement.capability_id);
208        let candidates = document
209            .capabilities
210            .iter()
211            .filter(|capability| {
212                capability.contract_id.starts_with(&prefix)
213                    && capability.contract_id == format!("{}{}", prefix, capability.major_version)
214            })
215            .collect::<Vec<_>>();
216        if candidates.is_empty() {
217            negotiation_issue(
218                &mut issues,
219                CapabilityNegotiationIssueCode::MissingCapability,
220                requirement,
221                "The managed Service does not advertise this capability",
222            );
223            continue;
224        }
225        let candidate = candidates
226            .into_iter()
227            .filter(|capability| {
228                requirement
229                    .supported_major_versions
230                    .contains(&capability.major_version)
231            })
232            .max_by_key(|capability| capability.major_version);
233        let Some(candidate) = candidate else {
234            negotiation_issue(
235                &mut issues,
236                CapabilityNegotiationIssueCode::UnsupportedMajorVersion,
237                requirement,
238                "The managed Service and consumer share no supported major version",
239            );
240            continue;
241        };
242        if !requirement
243            .required_feature_ids
244            .is_subset(&candidate.feature_ids)
245        {
246            negotiation_issue(
247                &mut issues,
248                CapabilityNegotiationIssueCode::MissingRequiredFeature,
249                requirement,
250                "The advertised contract is missing a required feature identifier",
251            );
252            continue;
253        }
254        if !requirement.accepted_schema_digests.is_empty()
255            && !requirement
256                .accepted_schema_digests
257                .contains(&candidate.schema_digest)
258        {
259            negotiation_issue(
260                &mut issues,
261                CapabilityNegotiationIssueCode::SchemaDigestMismatch,
262                requirement,
263                "The advertised schema digest is not accepted by the consumer",
264            );
265            continue;
266        }
267        accepted.push(NegotiatedCapability {
268            capability_id: requirement.capability_id.clone(),
269            advertisement: candidate.clone(),
270        });
271    }
272
273    accepted.sort_by(|left, right| left.capability_id.cmp(&right.capability_id));
274    CapabilityNegotiation { accepted, issues }
275}
276
277fn negotiation_issue(
278    issues: &mut Vec<CapabilityNegotiationIssue>,
279    code: CapabilityNegotiationIssueCode,
280    requirement: &CapabilityRequirement,
281    message: &str,
282) {
283    issues.push(CapabilityNegotiationIssue {
284        code,
285        capability_id: requirement.capability_id.clone(),
286        message: message.to_owned(),
287    });
288}
289
290fn valid_capability_id(value: &str) -> bool {
291    !value.is_empty()
292        && value.split('.').all(|segment| {
293            !segment.is_empty()
294                && !segment.starts_with('-')
295                && !segment.ends_with('-')
296                && !segment.contains("--")
297                && segment
298                    .bytes()
299                    .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-')
300        })
301}
302
303#[derive(Clone)]
304pub struct SystemPlaneAccess {
305    provider: Arc<dyn WorkloadIdentityProvider>,
306    audience: String,
307    enrollment_authorizer: Arc<dyn EnrollmentAuthorizer>,
308}
309
310impl fmt::Debug for SystemPlaneAccess {
311    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
312        formatter
313            .debug_struct("SystemPlaneAccess")
314            .field("provider", &self.provider)
315            .field("audience", &self.audience)
316            .field("enrollment_authorizer", &self.enrollment_authorizer)
317            .finish()
318    }
319}
320
321impl SystemPlaneAccess {
322    #[must_use]
323    pub fn new(
324        provider: Arc<dyn WorkloadIdentityProvider>,
325        audience: impl Into<String>,
326        enrollment_authorizer: Arc<dyn EnrollmentAuthorizer>,
327    ) -> Self {
328        Self {
329            provider,
330            audience: audience.into(),
331            enrollment_authorizer,
332        }
333    }
334}
335
336#[derive(Debug, Clone)]
337pub struct SystemPlaneRuntime {
338    pub registry: Arc<SystemPlaneRegistry>,
339    pub access: Arc<SystemPlaneAccess>,
340    pub module_operations: Option<Arc<ModuleOperationsProvider>>,
341}
342
343impl SystemPlaneRuntime {
344    #[must_use]
345    pub fn new(registry: SystemPlaneRegistry, access: SystemPlaneAccess) -> Self {
346        Self {
347            registry: Arc::new(registry),
348            access: Arc::new(access),
349            module_operations: None,
350        }
351    }
352
353    #[must_use]
354    pub fn with_module_operations(mut self, provider: ModuleOperationsProvider) -> Self {
355        self.module_operations = Some(Arc::new(provider));
356        self
357    }
358}
359
360#[derive(Debug, Serialize, ToSchema)]
361#[serde(rename_all = "camelCase")]
362pub struct SystemPlaneErrorBody {
363    pub code: &'static str,
364    pub message: String,
365    pub next_actions: Vec<&'static str>,
366}
367
368#[derive(Debug)]
369pub struct SystemPlaneRejection {
370    status: StatusCode,
371    code: &'static str,
372    message: String,
373    next_action: &'static str,
374}
375
376impl SystemPlaneRejection {
377    #[must_use]
378    pub fn new(
379        status: StatusCode,
380        code: &'static str,
381        message: impl Into<String>,
382        next_action: &'static str,
383    ) -> Self {
384        Self {
385            status,
386            code,
387            message: message.into(),
388            next_action,
389        }
390    }
391
392    #[must_use]
393    pub fn unavailable(
394        code: &'static str,
395        message: impl Into<String>,
396        next_action: &'static str,
397    ) -> Self {
398        Self {
399            status: StatusCode::SERVICE_UNAVAILABLE,
400            code,
401            message: message.into(),
402            next_action,
403        }
404    }
405
406    #[must_use]
407    pub const fn code(&self) -> &'static str {
408        self.code
409    }
410}
411
412impl IntoResponse for SystemPlaneRejection {
413    fn into_response(self) -> Response {
414        let body = SystemPlaneErrorBody {
415            code: self.code,
416            message: self.message,
417            next_actions: vec![self.next_action],
418        };
419        let mut response = (self.status, Json(body)).into_response();
420        response.headers_mut().insert(
421            header::CONTENT_TYPE,
422            HeaderValue::from_static("application/problem+json"),
423        );
424        response
425            .headers_mut()
426            .insert("x-lenso-error-code", HeaderValue::from_static(self.code));
427        response
428    }
429}
430
431#[derive(Debug, Clone)]
432pub struct AuthorizedSystemPlaneCaller {
433    pub runtime: Arc<SystemPlaneRuntime>,
434    pub service_principal: String,
435    pub enrollment: EnrollmentAuthorization,
436}
437
438impl AuthorizedSystemPlaneCaller {
439    pub fn require_capability(
440        &self,
441        contract_id: &str,
442        schema_digest: &str,
443        required_feature_ids: impl IntoIterator<Item = impl AsRef<str>>,
444    ) -> Result<(), SystemPlaneRejection> {
445        if self.enrollment.system_id == "system-sandbox" {
446            return Ok(());
447        }
448        let required_feature_ids = required_feature_ids
449            .into_iter()
450            .map(|feature| feature.as_ref().to_owned())
451            .collect::<BTreeSet<_>>();
452        let granted = self.enrollment.capabilities.iter().any(|capability| {
453            capability.contract_id == contract_id
454                && capability.schema_digest == schema_digest
455                && required_feature_ids.is_subset(&capability.feature_ids)
456        });
457        if granted {
458            Ok(())
459        } else {
460            Err(SystemPlaneRejection {
461                status: StatusCode::FORBIDDEN,
462                code: "system_plane_capability_not_granted",
463                message: "Active Enrollment Grant does not authorize the requested capability"
464                    .to_owned(),
465                next_action: "review_service_enrollment_grant",
466            })
467        }
468    }
469}
470
471impl<S> FromRequestParts<S> for AuthorizedSystemPlaneCaller
472where
473    S: Send + Sync,
474{
475    type Rejection = SystemPlaneRejection;
476
477    async fn from_request_parts(
478        parts: &mut axum::http::request::Parts,
479        _state: &S,
480    ) -> Result<Self, Self::Rejection> {
481        let runtime = parts
482            .extensions
483            .get::<Option<Arc<SystemPlaneRuntime>>>()
484            .and_then(Clone::clone)
485            .ok_or_else(|| SystemPlaneRejection {
486                status: StatusCode::SERVICE_UNAVAILABLE,
487                code: "system_plane_unavailable",
488                message: "System Plane access is not configured for this Service".to_owned(),
489                next_action: "configure_system_plane",
490            })?;
491        let token = bearer_token(&parts.headers)?;
492        let binding = parts
493            .extensions
494            .get::<AuthenticatedTransportBinding>()
495            .ok_or_else(|| SystemPlaneRejection {
496                status: StatusCode::UNAUTHORIZED,
497                code: "system_plane_transport_binding_required",
498                message: "System Plane access requires an authenticated transport binding"
499                    .to_owned(),
500                next_action: "use_authenticated_transport",
501            })?;
502        let principal = runtime
503            .access
504            .provider
505            .verify(
506                token,
507                &WorkloadIdentityVerification::new(
508                    &runtime.access.audience,
509                    &binding.0,
510                    now_unix_ms(),
511                ),
512            )
513            .map_err(|error| SystemPlaneRejection {
514                status: StatusCode::UNAUTHORIZED,
515                code: "system_plane_workload_identity_rejected",
516                message: error.message,
517                next_action: "refresh_workload_identity",
518            })?;
519        let enrollment = runtime
520            .access
521            .enrollment_authorizer
522            .authorize(
523                &runtime.registry.document().service_id,
524                &principal.service_principal,
525                now_unix_ms(),
526            )
527            .await
528            .map_err(enrollment_rejection)?;
529        Ok(Self {
530            runtime,
531            service_principal: principal.service_principal,
532            enrollment,
533        })
534    }
535}
536
537/// Builds the mandatory Core discovery route. An absent runtime fails closed.
538#[must_use]
539pub fn router<S>(runtime: Option<Arc<SystemPlaneRuntime>>) -> OpenApiRouter<S>
540where
541    S: Clone + Send + Sync + 'static,
542{
543    OpenApiRouter::new()
544        .routes(routes!(discover_core))
545        .layer(Extension(runtime))
546}
547
548#[utoipa::path(
549    get,
550    path = "/system-plane/v1",
551    responses(
552        (status = 200, description = "Authenticated System Plane Core document", body = CoreDocument),
553        (status = 401, description = "Workload Identity or transport binding was not accepted", body = SystemPlaneErrorBody, content_type = "application/problem+json"),
554        (status = 403, description = "Caller is not the enrolled Console Service Principal", body = SystemPlaneErrorBody, content_type = "application/problem+json"),
555        (status = 503, description = "System Plane access is not configured", body = SystemPlaneErrorBody, content_type = "application/problem+json")
556    ),
557    security(("bearer_auth" = [])),
558    tag = "system-plane"
559)]
560async fn discover_core(caller: AuthorizedSystemPlaneCaller) -> Json<CoreDocument> {
561    Json(caller.runtime.registry.document().clone())
562}
563
564fn bearer_token(headers: &HeaderMap) -> Result<&str, SystemPlaneRejection> {
565    headers
566        .get(header::AUTHORIZATION)
567        .and_then(|value| value.to_str().ok())
568        .and_then(|value| value.strip_prefix("Bearer "))
569        .filter(|value| !value.is_empty())
570        .ok_or_else(|| SystemPlaneRejection {
571            status: StatusCode::UNAUTHORIZED,
572            code: "system_plane_workload_identity_required",
573            message: "System Plane access requires a Workload Identity Bearer credential"
574                .to_owned(),
575            next_action: "provide_workload_identity",
576        })
577}
578
579fn enrollment_rejection(error: EnrollmentError) -> SystemPlaneRejection {
580    let (status, code, next_action) = match error.code {
581        EnrollmentErrorCode::StoreUnavailable => (
582            StatusCode::SERVICE_UNAVAILABLE,
583            "system_plane_enrollment_unavailable",
584            "restore_enrollment_store",
585        ),
586        EnrollmentErrorCode::Expired => (
587            StatusCode::FORBIDDEN,
588            "system_plane_enrollment_expired",
589            "renew_service_enrollment",
590        ),
591        EnrollmentErrorCode::Revoked => (
592            StatusCode::FORBIDDEN,
593            "system_plane_enrollment_revoked",
594            "complete_service_enrollment",
595        ),
596        EnrollmentErrorCode::PrincipalMismatch => (
597            StatusCode::FORBIDDEN,
598            "system_plane_console_not_enrolled",
599            "complete_service_enrollment",
600        ),
601        EnrollmentErrorCode::NotEnrolled
602        | EnrollmentErrorCode::InvalidGrant
603        | EnrollmentErrorCode::InvalidDecision
604        | EnrollmentErrorCode::SignatureRejected
605        | EnrollmentErrorCode::NonceReused
606        | EnrollmentErrorCode::AlreadyEnrolled
607        | EnrollmentErrorCode::StaleAuthorizationEpoch => (
608            StatusCode::FORBIDDEN,
609            "system_plane_enrollment_required",
610            "complete_service_enrollment",
611        ),
612    };
613    SystemPlaneRejection {
614        status,
615        code,
616        message: error.message,
617        next_action,
618    }
619}
620
621fn now_unix_ms() -> u64 {
622    SystemTime::now()
623        .duration_since(UNIX_EPOCH)
624        .map_or(0, |duration| {
625            u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
626        })
627}
platform_system_plane/lib.rs

platform_system_plane/
lib.rs
