systemprompt_loader/bundle/source/oci/
push.rs

1//! Publishing a bundle to an OCI registry.
2//!
3//! Uploads are monolithic: a session is opened, the whole blob is PUT with
4//! its digest, and the registry's own digest check is the acceptance test.
5//! A push whose response the registry does not accept is an error — a
6//! publish that "mostly worked" would leave a manifest pointing at bytes no
7//! instance can pull.
8//!
9//! Copyright (c) systemprompt.io — Business Source License 1.1.
10//! See <https://systemprompt.io> for licensing details.
11
12use std::path::Path;
13
14use sha2::{Digest, Sha256};
15use systemprompt_models::services::bundle::BUNDLE_MEDIA_TYPE;
16
17use super::RegistryClient;
18use super::pull::{OCI_MANIFEST_MEDIA_TYPE, OciDescriptor, OciManifest};
19use crate::bundle::error::{BundleError, BundleResult};
20
21pub const BUNDLE_CONFIG_MEDIA_TYPE: &str =
22    "application/vnd.systemprompt.services-bundle.config.v1+json";
23
24pub async fn push_bundle(
25    reference: &str,
26    archive: &Path,
27    manifest_json: &[u8],
28    secret: Option<String>,
29    client: reqwest::Client,
30) -> BundleResult<String> {
31    let registry = RegistryClient::new("publish", reference, secret, client)?;
32    let archive_bytes = tokio::fs::read(archive).await?;
33
34    let config = upload_blob(&registry, manifest_json, BUNDLE_CONFIG_MEDIA_TYPE).await?;
35    let layer = upload_blob(&registry, &archive_bytes, BUNDLE_MEDIA_TYPE).await?;
36
37    let manifest = OciManifest {
38        schema_version: 2,
39        media_type: Some(OCI_MANIFEST_MEDIA_TYPE.to_owned()),
40        artifact_type: Some(BUNDLE_MEDIA_TYPE.to_owned()),
41        config,
42        layers: vec![layer],
43    };
44    put_manifest(&registry, &manifest).await
45}
46
47async fn upload_blob(
48    registry: &RegistryClient,
49    body: &[u8],
50    media_type: &str,
51) -> BundleResult<OciDescriptor> {
52    let digest = format!("sha256:{}", hex::encode(Sha256::digest(body)));
53
54    let initiate_url = registry.url("/blobs/uploads/")?;
55    let initiated = registry
56        .send(move |client| client.post(initiate_url.clone()))
57        .await?;
58    if !initiated.status().is_success() {
59        return Err(BundleError::fetch(
60            &registry.name,
61            format!("upload session refused: {}", initiated.status()),
62        ));
63    }
64    let location = initiated
65        .headers()
66        .get(reqwest::header::LOCATION)
67        .and_then(|v| v.to_str().ok())
68        .ok_or_else(|| BundleError::fetch(&registry.name, "upload session has no Location"))?
69        .to_owned();
70    let upload_url = absolute_location(registry, &location)?;
71
72    let owned = body.to_vec();
73    let digest_param = digest.clone();
74    let response = registry
75        .send(move |client| {
76            client
77                .put(upload_url.clone())
78                .query(&[("digest", digest_param.as_str())])
79                .header(reqwest::header::CONTENT_TYPE, "application/octet-stream")
80                .body(owned.clone())
81        })
82        .await?;
83    if !response.status().is_success() {
84        return Err(BundleError::fetch(
85            &registry.name,
86            format!("blob upload failed: {}", response.status()),
87        ));
88    }
89
90    Ok(OciDescriptor {
91        media_type: media_type.to_owned(),
92        digest,
93        size: body.len() as u64,
94    })
95}
96
97fn absolute_location(registry: &RegistryClient, location: &str) -> BundleResult<url::Url> {
98    if location.starts_with("http://") || location.starts_with("https://") {
99        return url::Url::parse(location)
100            .map_err(|e| BundleError::fetch(&registry.name, format!("bad upload location: {e}")));
101    }
102    let base = registry.url("")?;
103    base.join(location)
104        .map_err(|e| BundleError::fetch(&registry.name, format!("bad upload location: {e}")))
105}
106
107async fn put_manifest(registry: &RegistryClient, manifest: &OciManifest) -> BundleResult<String> {
108    let body = serde_json::to_vec(manifest)
109        .map_err(|e| BundleError::policy(format!("manifest is not serialisable: {e}")))?;
110    let digest = format!("sha256:{}", hex::encode(Sha256::digest(&body)));
111    let url = registry.url(&format!("/manifests/{}", registry.manifest_ref()))?;
112
113    let response = registry
114        .send(move |client| {
115            client
116                .put(url.clone())
117                .header(reqwest::header::CONTENT_TYPE, OCI_MANIFEST_MEDIA_TYPE)
118                .body(body.clone())
119        })
120        .await?;
121    if !response.status().is_success() {
122        return Err(BundleError::fetch(
123            &registry.name,
124            format!("manifest push failed: {}", response.status()),
125        ));
126    }
127    Ok(digest)
128}
systemprompt_loader/bundle/source/oci/push.rs

systemprompt_loader/bundle/source/oci/
push.rs
