ianoc · Oct 29, 2024 · Oct 29, 2024
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -36,6 +36,7 @@ http = "0.2.11"
 hyper = { version = "0.14.27", features = ["full"] }
 hyper-rustls = "0.23.2"
 indicatif = { git = "https://github.com/console-rs/indicatif.git", rev = "2ca9d019fbb4b0aed11bfad984daef064998f5ef" }
+regex = "1.10.3"
 rustls = "0.20.8"
 serde = { version = "1.0.198", features = ["derive", "alloc"] }
 serde_json = { version = "1.0.116", features = ["alloc"] }

diff --git a/app/puller_app.rs b/app/puller_app.rs
@@ -1,5 +1,6 @@
-use std::path::PathBuf;
+use std::{path::PathBuf, sync::Arc};
 
+use anyhow::Context;
 use clap::Parser;
 
 use rules_minidock_tools::container_specs::{ConfigDelta, Manifest};
@@ -20,15 +21,30 @@ struct Opt {
 
     #[clap(long)]
     architecture: String,
+
+    #[clap(long)]
+    // List of comma separated helpers. with the registry:helper_path
+    //e.g. foo.gcr.io:/path/to/helper,bar.gcr.io:/path/to/helper2
+    docker_authorization_helpers: Option<String>,
 }
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
     let opt = Opt::parse();
 
-    let registry =
-        rules_minidock_tools::registry::from_maybe_domain_and_name(&opt.registry, &opt.repository)
-            .await?;
+    let docker_authorization_helpers = if let Some(arg) = &opt.docker_authorization_helpers {
+        Arc::new(rules_minidock_tools::registry::DockerAuthenticationHelper::from_str(arg)?)
+    } else {
+        Default::default()
+    };
+
+    let registry = rules_minidock_tools::registry::from_maybe_domain_and_name(
+        &opt.registry,
+        &opt.repository,
+        docker_authorization_helpers,
+    )
+    .await
+    .with_context(|| format!("Failed to connect to registry name: {}", opt.registry))?;
     let manifest_ret = registry.fetch_manifest_as_string(&opt.digest).await?;
 
     let cfg_path = PathBuf::from("config.json");

diff --git a/app/pusher_app.rs b/app/pusher_app.rs
@@ -36,6 +36,11 @@ struct Opt {
     /// Do not upload the manifests, meaning just blobs that don't effect tags/usage are synchronized.
     #[clap(long)]
     skip_manifest_upload: bool,
+
+    #[clap(long)]
+    // List of comma separated helpers. with the registry:helper_path
+    //e.g. foo.gcr.io:/path/to/helper,bar.gcr.io:/path/to/helper2
+    docker_authorization_helpers: Option<String>,
 }
 
 #[derive(Deserialize, Serialize, Debug, PartialEq, Eq, Clone)]
@@ -94,6 +99,12 @@ async fn main() -> Result<(), anyhow::Error> {
         );
     }
 
+    let docker_authorization_helpers = if let Some(arg) = &opt.docker_authorization_helpers {
+        Arc::new(rules_minidock_tools::registry::DockerAuthenticationHelper::from_str(arg)?)
+    } else {
+        Default::default()
+    };
+
     let pusher_config_content = std::fs::read_to_string(&opt.pusher_config)?;
     let pusher_config: PusherConfig = serde_json::from_str(pusher_config_content.as_str())
         .with_context(|| {
@@ -130,8 +141,14 @@ async fn main() -> Result<(), anyhow::Error> {
                 bail!("Passed in an invalid registry, its an empty string.")
             }
             let repository = pusher_config.repository.clone();
+            let docker_authorization_helpers = docker_authorization_helpers.clone();
             Ok(tokio::spawn(async move {
-                rules_minidock_tools::registry::from_maybe_domain_and_name(&r, &repository).await
+                rules_minidock_tools::registry::from_maybe_domain_and_name(
+                    &r,
+                    &repository,
+                    docker_authorization_helpers,
+                )
+                .await
             }))
         })
         .collect();
@@ -158,8 +175,12 @@ async fn main() -> Result<(), anyhow::Error> {
                 None
             }
             (Some(registry), Some(repository)) => Some(
-                rules_minidock_tools::registry::from_maybe_domain_and_name(&registry, &repository)
-                    .await?,
+                rules_minidock_tools::registry::from_maybe_domain_and_name(
+                    &registry,
+                    &repository,
+                    docker_authorization_helpers,
+                )
+                .await?,
             ),
         }
     } else {

diff --git a/src/container_specs/manifest.rs b/src/container_specs/manifest.rs
@@ -18,7 +18,7 @@ pub struct Manifest {
 impl Manifest {
     pub fn media_type(&self) -> &'static str {
         match self.specification_type {
-            SpecificationType::Oci => "application/vnd.oci.image.config.v1+json",
+            SpecificationType::Oci => "application/vnd.oci.image.manifest.v1+json",
             SpecificationType::Docker => "application/vnd.docker.distribution.manifest.v2+json",
         }
     }

diff --git a/src/container_specs/serde_impl.rs b/src/container_specs/serde_impl.rs
@@ -39,7 +39,12 @@ impl<'de> Deserialize<'de> for BlobReference {
             "application/vnd.docker.image.rootfs.diff.tar" => {
                 (SpecificationType::Docker, BlobReferenceType::Layer)
             }
-            other => return Err(D::Error::custom(format!("Invalid media type: {}", other))),
+            other => {
+                return Err(D::Error::custom(format!(
+                    "Invalid media type for BlobReference: {}",
+                    other
+                )))
+            }
         };
 
         Ok(BlobReference {
@@ -112,9 +117,14 @@ impl<'de> Deserialize<'de> for Manifest {
         }
 
         let specification_type = match r.media_type.as_str() {
-            "application/vnd.oci.image.config.v1+json" => SpecificationType::Oci,
+            "application/vnd.oci.image.manifest.v1+json" => SpecificationType::Oci,
             "application/vnd.docker.distribution.manifest.v2+json" => SpecificationType::Docker,
-            other => return Err(D::Error::custom(format!("Invalid media type: {}", other))),
+            other => {
+                return Err(D::Error::custom(format!(
+                    "Invalid media type for Manifest: {}",
+                    other
+                )))
+            }
         };
 
         Ok(Manifest {

diff --git a/src/registry/http/blob.rs b/src/registry/http/blob.rs
@@ -143,6 +143,21 @@ impl BlobStore for super::HttpRegistry {
             bail!("Was a redirection response code, but missing Location header, invalid response from server, body:\n{:#?}", body);
         };
 
+        // Sometimes we can receive new URI's that don't contain hosts
+        // we need to supply this information from the last URI we used in that case
+        let location_uri = if location_uri.host().is_some() {
+            location_uri
+        } else {
+            let mut parts = post_target_uri.into_parts();
+            parts.path_and_query = location_uri.path_and_query().cloned();
+            Uri::from_parts(parts).with_context(|| {
+                format!(
+                    "Constructed an invalid uri from parts, new uri: {:?}",
+                    location_uri
+                )
+            })?
+        };
+
         let total_uploaded_bytes = Arc::new(Mutex::new(0));
 
         struct Context {

diff --git a/src/registry/http/http_cli/authentication_flow.rs b/src/registry/http/http_cli/authentication_flow.rs
@@ -1,4 +1,6 @@
-use crate::registry::http::util::dump_body_to_string;
+use std::sync::Arc;
+
+use crate::registry::{http::util::dump_body_to_string, DockerAuthenticationHelper};
 
 use anyhow::Context;
 
@@ -7,6 +9,7 @@ use http::Uri;
 use hyper::{Body, Client};
 
 use serde::{Deserialize, Serialize};
+use tokio::{io::AsyncWriteExt, process::Command};
 
 use super::private_impl::{run_single_request, BearerConfig};
 
@@ -18,9 +21,19 @@ pub struct AuthResponse {
     pub issued_at: Option<String>,
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+struct BasicAuthResponse {
+    #[serde(rename = "Username")]
+    username: String,
+
+    #[serde(rename = "Secret")]
+    secret: String,
+}
+
 pub async fn authenticate_request(
     auth_fail: &BearerConfig,
     inner_client: &Client<hyper_rustls::HttpsConnector<hyper::client::HttpConnector>>,
+    docker_authorization_helpers: Arc<Vec<DockerAuthenticationHelper>>,
 ) -> anyhow::Result<AuthResponse> {
     let mut parts = auth_fail.realm.clone().into_parts();
     let new_query_items = if let Some(scope) = &auth_fail.scope {
@@ -50,9 +63,67 @@ pub async fn authenticate_request(
             new_path_q
         )
     })?;
-    // The wiring to supply basic_auth_info is TBD next
-    // right now it can never be set. Need to use the docker credentials helpers to supply this.
-    let basic_auth_info: Option<String> = None;
+
+    let matching_helper_opt: Option<&DockerAuthenticationHelper> = docker_authorization_helpers
+        .iter()
+        .find(|e| e.registry == auth_fail.service);
+
+    let basic_auth_info = if let Some(matching_helper) = matching_helper_opt {
+        let mut child = Command::new(&matching_helper.helper_path)
+            .stdin(std::process::Stdio::piped())
+            .stdout(std::process::Stdio::piped())
+            .spawn()
+            .with_context(|| {
+                format!(
+                    "Failed to start helper program at {:?}",
+                    matching_helper.helper_path
+                )
+            })?;
+
+        let mut child_stdin = child
+            .stdin
+            .take()
+            .ok_or(anyhow::anyhow!("Failed to get stdin from running process"))?;
+        child_stdin
+            .write_all(format!("GET {}\n", &auth_fail.service).as_bytes())
+            .await
+            .with_context(|| {
+                format!(
+                    "Failed when trying to send domain into the stdin of the auth helper at {:?}",
+                    &matching_helper.helper_path
+                )
+            })?;
+        drop(child_stdin);
+
+        let output = child.wait_with_output().await.with_context(|| {
+            format!(
+                "Failed waiting for output from subprocess calling {:?}",
+                matching_helper.helper_path
+            )
+        })?;
+
+        if output.status.success() {
+            let stdout_str = String::from_utf8_lossy(&output.stdout);
+            Some(
+                serde_json::from_str::<BasicAuthResponse>(&stdout_str).with_context(|| {
+                    format!(
+                        "Failed to parse output from {:?}, saw output:\n{}",
+                        matching_helper.helper_path, stdout_str
+                    )
+                })?,
+            )
+        } else {
+            anyhow::bail!(
+                "Failed to run helper program at {:?}, got status code: {:?}, stderr: {:?}",
+                matching_helper.helper_path,
+                output.status,
+                String::from_utf8(output.stderr)
+            );
+        }
+    } else {
+        None
+    };
+
     let mut response = run_single_request(
         Default::default(),
         &new_uri,
@@ -64,7 +135,10 @@ pub async fn authenticate_request(
             let b3 = if let Some(ai) = basic_auth_info {
                 b2.header(
                     "Authorization",
-                    format!("Basic {}", BASE64_STANDARD.encode(ai)),
+                    format!(
+                        "Basic {}",
+                        BASE64_STANDARD.encode(format!("{}:{}", ai.username, ai.secret).as_bytes())
+                    ),
                 )
             } else {
                 b2