systemprompt_loader/bundle/source/oci/
pull.rs

1//! Manifest and blob reads against an OCI registry.
2//!
3//! Registries answer a blob GET with a redirect to their storage backend
4//! (GHCR: `307` to `pkg-containers.githubusercontent.com`), so the blob read
5//! follows a bounded chain of redirects itself — the shared client is built
6//! without redirect following so the registry credential never travels to a
7//! host the profile did not name. The redirected request is sent bare: the
8//! target URL carries its own signed authorisation, and forwarding the
9//! registry token to a CDN would leak it.
10//!
11//! Copyright (c) systemprompt.io — Business Source License 1.1.
12//! See <https://systemprompt.io> for licensing details.
13
14use std::path::Path;
15
16use serde::{Deserialize, Serialize};
17use sha2::{Digest, Sha256};
18use systemprompt_models::services::bundle::BUNDLE_MEDIA_TYPE;
19
20use super::RegistryClient;
21use crate::bundle::error::{BundleError, BundleResult};
22use crate::bundle::source::FetchedBundle;
23use crate::bundle::source::stream::stream_to_file;
24use systemprompt_models::net::{trusted_http_hosts_from_env, validate_outbound_url_with_trust};
25
26pub const OCI_MANIFEST_MEDIA_TYPE: &str = "application/vnd.oci.image.manifest.v1+json";
27pub const DOCKER_MANIFEST_MEDIA_TYPE: &str = "application/vnd.docker.distribution.manifest.v2+json";
28
29#[derive(Debug, Clone, Serialize, Deserialize)]
30pub struct OciDescriptor {
31    #[serde(rename = "mediaType")]
32    pub media_type: String,
33
34    pub digest: String,
35
36    pub size: u64,
37}
38
39#[derive(Debug, Clone, Serialize, Deserialize)]
40pub struct OciManifest {
41    #[serde(rename = "schemaVersion")]
42    pub schema_version: u32,
43
44    #[serde(rename = "mediaType", default, skip_serializing_if = "Option::is_none")]
45    pub media_type: Option<String>,
46
47    #[serde(
48        rename = "artifactType",
49        default,
50        skip_serializing_if = "Option::is_none"
51    )]
52    pub artifact_type: Option<String>,
53
54    pub config: OciDescriptor,
55
56    #[serde(default)]
57    pub layers: Vec<OciDescriptor>,
58}
59
60pub async fn get_manifest(registry: &RegistryClient) -> BundleResult<(String, OciManifest)> {
61    let url = registry.url(&format!("/manifests/{}", registry.manifest_ref()))?;
62    let accept = format!("{OCI_MANIFEST_MEDIA_TYPE}, {DOCKER_MANIFEST_MEDIA_TYPE}");
63    let response = registry
64        .send(move |client| {
65            client
66                .get(url.clone())
67                .header(reqwest::header::ACCEPT, accept.clone())
68        })
69        .await?;
70
71    let status = response.status();
72    if !status.is_success() {
73        return Err(BundleError::fetch(
74            &registry.name,
75            format!("manifest request failed: {status}"),
76        ));
77    }
78
79    let header_digest = response
80        .headers()
81        .get("docker-content-digest")
82        .and_then(|v| v.to_str().ok())
83        .map(str::to_owned);
84    let body = response
85        .bytes()
86        .await
87        .map_err(|e| BundleError::fetch(&registry.name, e))?;
88    let digest =
89        header_digest.unwrap_or_else(|| format!("sha256:{}", hex::encode(Sha256::digest(&body))));
90
91    let manifest: OciManifest = serde_json::from_slice(&body)
92        .map_err(|e| BundleError::fetch(&registry.name, format!("manifest does not parse: {e}")))?;
93    Ok((digest, manifest))
94}
95
96pub async fn pull_bundle_layer(
97    registry: &RegistryClient,
98    manifest: &OciManifest,
99    into: &Path,
100    max_bytes: u64,
101) -> BundleResult<FetchedBundle> {
102    let matching: Vec<&OciDescriptor> = manifest
103        .layers
104        .iter()
105        .filter(|l| l.media_type == BUNDLE_MEDIA_TYPE)
106        .collect();
107    let [layer] = matching.as_slice() else {
108        return Err(BundleError::fetch(
109            &registry.name,
110            format!(
111                "manifest carries {} layers of {BUNDLE_MEDIA_TYPE}, expected exactly one",
112                matching.len()
113            ),
114        ));
115    };
116
117    if layer.size > max_bytes {
118        return Err(BundleError::TooLarge { bytes: max_bytes });
119    }
120
121    let url = registry.url(&format!("/blobs/{}", layer.digest))?;
122    let response = registry.send(move |client| client.get(url.clone())).await?;
123    let response = follow_blob_redirects(registry, response).await?;
124    let status = response.status();
125    if !status.is_success() {
126        return Err(BundleError::fetch(
127            &registry.name,
128            format!("blob request failed: {status}"),
129        ));
130    }
131
132    let digest = stream_to_file(response, into, &registry.name, max_bytes).await?;
133    let expected = layer
134        .digest
135        .strip_prefix("sha256:")
136        .unwrap_or(&layer.digest);
137    if !digest.eq_ignore_ascii_case(expected) {
138        return Err(BundleError::fetch(
139            &registry.name,
140            format!(
141                "blob digest is sha256:{digest}, manifest declares {}",
142                layer.digest
143            ),
144        ));
145    }
146
147    Ok(FetchedBundle {
148        archive: into.to_path_buf(),
149        digest: format!("sha256:{digest}"),
150    })
151}
152
153// Why: bounded so a registry that redirects in a loop is a fetch error, not a
154// hang; three hops covers every known registry → CDN → signed-URL chain.
155const MAX_BLOB_REDIRECTS: usize = 3;
156
157async fn follow_blob_redirects(
158    registry: &RegistryClient,
159    mut response: reqwest::Response,
160) -> BundleResult<reqwest::Response> {
161    for _ in 0..MAX_BLOB_REDIRECTS {
162        if !response.status().is_redirection() {
163            return Ok(response);
164        }
165        let location = response
166            .headers()
167            .get(reqwest::header::LOCATION)
168            .and_then(|v| v.to_str().ok())
169            .ok_or_else(|| {
170                BundleError::fetch(
171                    &registry.name,
172                    format!("blob redirect ({}) without a Location", response.status()),
173                )
174            })?;
175        let target = match url::Url::parse(location) {
176            Ok(absolute) => absolute,
177            Err(_) => response
178                .url()
179                .join(location)
180                .map_err(|e| BundleError::fetch(&registry.name, format!("blob redirect: {e}")))?,
181        };
182        let target =
183            validate_outbound_url_with_trust(target.as_str(), &trusted_http_hosts_from_env())
184                .map_err(|e| BundleError::fetch(&registry.name, e))?;
185        response = registry
186            .client
187            .get(target)
188            .send()
189            .await
190            .map_err(|e| BundleError::fetch(&registry.name, e))?;
191    }
192    Err(BundleError::fetch(
193        &registry.name,
194        format!("blob redirected more than {MAX_BLOB_REDIRECTS} times"),
195    ))
196}
systemprompt_loader/bundle/source/oci/pull.rs

systemprompt_loader/bundle/source/oci/
pull.rs
