platform_system_plane/
lib.rs

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

platform_system_plane/
lib.rs
