Merge pull request #233 from flavio/fix-http-registry-interactions

fix: http registry interactions

.taplo.toml

@@ -0,0 +1,5 @@
+[formatting]
+align_entries  = true
+reorder_arrays = true
+reorder_keys   = true
+sort_keys      = true

Cargo.toml

@@ -1,59 +1,61 @@
 [package]
-name = "policy-fetcher"
-version = "0.9.0"
 authors = [
-  "Kubewarden Developers <kubewarden@suse.de>",
+  "Fabrizio Sestito <fabrizio.sestito@suse.com>",
   "Flavio Castelli <fcastelli@suse.com>",
+  "Kubewarden Developers <kubewarden@suse.de>",
   "Rafael Fernández López <rfernandezlopez@suse.com>",
   "Víctor Cuadrado Juan <vcuadradojuan@suse.de>",
-  "Fabrizio Sestito <fabrizio.sestito@suse.com>",
 ]
 edition = "2021"
+name = "policy-fetcher"
+version = "0.10.0"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 async-trait = "0.1"
 base64 = "0.22"
+cfg-if = "1.0"
 directories = "6.0"
+docker_credential = "1.2"
+futures = "0.3"
 lazy_static = "1.4"
 oci-client = { version = "0.14", default-features = false, features = [
   "rustls-tls",
 ] }
 path-slash = "0.2"
+rayon = "1.5"
 regex = "1.5"
 reqwest = { version = "0.12", default-features = false, features = [
   "rustls-tls",
 ] }
 rustls = { version = "0.23", default-features = false, features = [
-  "std",
   "ring",
+  "std",
   "tls12",
 ] }
 rustls-pki-types = "1.0.1" # stick to the same version used by sigstore
 serde = { version = "1.0", features = ["derive"] }
+serde_bytes = "0.11"
 serde_json = "1.0"
 serde_yaml = "0.9"
-serde_bytes = "0.11"
 sha2 = "0.10"
 sigstore = { version = "0.11.0", default-features = false, features = [
-  "sigstore-trust-root-rustls-tls",
-  "cosign-rustls-tls",
   "cached-client",
+  "cosign-rustls-tls",
+  "sigstore-trust-root-rustls-tls",
 ] }
 thiserror = "2.0"
+tokio = { version = "1", default-features = false }
 tracing = "0.1"
 url = { version = "2.2", features = ["serde"] }
 walkdir = "2.5"
-rayon = "1.5"
-docker_credential = "1.2"
-tokio = { version = "1", default-features = false }
-cfg-if = "1.0"
 x509-parser = "0.17"
 
 [dev-dependencies]
-rstest = "0.24"
-tempfile = "3.2"
-textwrap = "0.16"
+anyhow         = "1.0"
+rcgen          = "0.13"
+rstest         = "0.24"
+tempfile       = "3.2"
 testcontainers = { version = "0.23", features = ["http_wait"] }
-anyhow = "1.0"
+textwrap       = "0.16"

src/registry/mod.rs

@@ -1,5 +1,9 @@
+use std::{collections::BTreeMap, convert::TryFrom, str::FromStr};
+
 use async_trait::async_trait;
 use docker_credential::DockerCredential;
+use errors::RegistryError;
+use futures::future::BoxFuture;
 use lazy_static::lazy_static;
 use oci_client::{
     client::{
@@ -11,22 +15,14 @@ use oci_client::{
     Reference,
 };
 use regex::Regex;
-use std::convert::TryFrom;
-use std::str::FromStr;
 use tracing::{debug, info, warn};
 use url::Url;
 
 use crate::{
-    fetcher::TlsVerificationMode,
-    sources::{Certificate, SourceResult, Sources},
-};
-use crate::{
-    fetcher::{ClientProtocol, PolicyFetcher},
-    sources::SourceError,
+    fetcher::{ClientProtocol, PolicyFetcher, TlsVerificationMode},
+    registry::errors::RegistryResult,
+    sources::{Certificate, SourceError, SourceResult, Sources},
 };
-use errors::RegistryError;
-
-use self::errors::RegistryResult;
 
 pub mod errors;
 
@@ -93,6 +89,45 @@ impl From<ClientProtocol> for ClientConfig {
     }
 }
 
+/// Perform the given operation using different protocols. This is done because we want to use
+/// following behavior:
+/// - Try with HTTPS, using either the system CA certificates or the custom ones provided by the user
+/// - If the connection fails, check if the destination was marked as insecure. If that's the case,
+///   try again, this time disabling TLS verification
+/// - If the connection still fails, try one last time, this time using HTTP instead of HTTPS
+async fn try_with_protocols<'a, F, T>(
+    url: &'a Url,
+    sources: &'a Sources,
+    operation: F,
+) -> RegistryResult<T>
+where
+    F: Fn(ClientProtocol) -> BoxFuture<'a, RegistryResult<T>>,
+{
+    let client_protocol = crate::client_protocol(url, sources)?;
+    match operation(client_protocol.clone()).await {
+        Ok(result) => return Ok(result),
+        Err(err) => {
+            if !sources.is_insecure_source(&crate::host_and_port(url)?) {
+                return Err(err);
+            }
+            info!(%err, %client_protocol, insecure_source = true, "operation failed");
+        }
+    }
+
+    // try again, this time disabling TLS verification because this source is marked as insecure
+    let client_protocol = ClientProtocol::Https(TlsVerificationMode::NoTlsVerification);
+    match operation(client_protocol.clone()).await {
+        Ok(result) => return Ok(result),
+        Err(err) => {
+            info!(%err, %client_protocol, insecure_source = true, "operation failed");
+        }
+    }
+
+    // try one last time, this time using HTTP instead of HTTPS. That because this source is marked as insecure
+    let client_protocol = ClientProtocol::Http;
+    operation(client_protocol).await
+}
+
 impl Registry {
     pub fn new() -> Registry {
         Registry {}
@@ -137,13 +172,22 @@ impl Registry {
         let url: Url = Url::parse(format!("registry://{}", reference).as_str())?;
         let registry_auth = Registry::auth(reference.registry());
         let sources: Sources = sources.cloned().unwrap_or_default();
-        let cp = crate::client_protocol(&url, &sources)?;
 
-        let (m, _) = Registry::client(cp)
-            .pull_manifest(&reference, &registry_auth)
-            .await?;
+        let (oci_manifest, _) = try_with_protocols(&url, &sources, |client_protocol| {
+            Box::pin({
+                let reference = reference.clone();
+                let registry_auth = registry_auth.clone();
+                async move {
+                    let res = Registry::client(client_protocol)
+                        .pull_manifest(&reference, &registry_auth)
+                        .await?;
+                    Ok(res)
+                }
+            })
+        })
+        .await?;
 
-        Ok(m)
+        Ok(oci_manifest)
     }
 
     /// Fetch the manifest's digest of the OCI object referenced by the given url.
@@ -159,11 +203,22 @@ impl Registry {
         let url: Url = Url::parse(format!("registry://{}", reference).as_str())?;
         let registry_auth = Registry::auth(reference.registry());
         let sources: Sources = sources.cloned().unwrap_or_default();
-        let cp = crate::client_protocol(&url, &sources)?;
 
-        Ok(Registry::client(cp)
-            .fetch_manifest_digest(&reference, &registry_auth)
-            .await?)
+        let digest = try_with_protocols(&url, &sources, |client_protocol| {
+            Box::pin({
+                let reference = reference.clone();
+                let registry_auth = registry_auth.clone();
+                async move {
+                    let res = Registry::client(client_protocol)
+                        .fetch_manifest_digest(&reference, &registry_auth)
+                        .await?;
+                    Ok(res)
+                }
+            })
+        })
+        .await?;
+
+        Ok(digest)
     }
 
     /// Push the policy to the OCI registry specified by `url`.
@@ -175,6 +230,7 @@ impl Registry {
         policy: &[u8],
         destination: &str,
         sources: Option<&Sources>,
+        annotations: Option<BTreeMap<String, String>>,
     ) -> RegistryResult<String> {
         let url = Url::parse(destination)
             .map_err(|_| crate::errors::InvalidURLError(destination.to_owned()))?;
@@ -183,36 +239,27 @@ impl Registry {
             .strip_prefix("registry://")
             .ok_or_else(|| RegistryError::InvalidDestinationError)?;
 
-        let client_protocol = crate::client_protocol(&url, &sources)?;
-        match self.do_push(policy, &url, client_protocol.clone()).await {
-            Ok(manifest_url) => {
-                return build_immutable_ref(destination, &manifest_url);
-            }
-            Err(err) => {
-                if !sources.is_insecure_source(&crate::host_and_port(&url)?) {
-                    return Err(err);
+        let manifest_url = try_with_protocols(&url.clone(), &sources, |client_protocol| {
+            Box::pin({
+                let url = url.clone();
+                let annotations = annotations.clone();
+                async move {
+                    let res = self
+                        .do_push(policy, &url, annotations.as_ref(), client_protocol.clone())
+                        .await?;
+                    Ok(res)
                 }
-                info!(%err, %client_protocol, insecure_source = true, "push failed");
-            }
-        }
-
-        let client_protocol = ClientProtocol::Https(TlsVerificationMode::NoTlsVerification);
-        match self.do_push(policy, &url, client_protocol.clone()).await {
-            Ok(manifest_url) => return build_immutable_ref(destination, &manifest_url),
-            Err(err) => {
-                info!(%err, %client_protocol, insecure_source = true, "push failed");
-            }
-        }
-
-        let manifest_url = self.do_push(policy, &url, ClientProtocol::Http).await?;
-
+            })
+        })
+        .await?;
         build_immutable_ref(destination, &manifest_url)
     }
 
     async fn do_push(
         &self,
         policy: &[u8],
         url: &Url,
+        annotations: Option<&BTreeMap<String, String>>,
         client_protocol: ClientProtocol,
     ) -> RegistryResult<String> {
         warn!(client_protocol = ?client_protocol, "pushing policy");
@@ -233,7 +280,9 @@ impl Registry {
             annotations: None,
         };
 
-        let image_manifest = manifest::OciImageManifest::build(&layers, &config, None);
+        let image_manifest =
+            manifest::OciImageManifest::build(&layers, &config, annotations.cloned());
+
         Ok(Registry::client(client_protocol)
             .push(
                 &reference,
@@ -245,6 +294,7 @@ impl Registry {
             .await
             .map(|push_response| push_response.manifest_url)?)
     }
+
     /// Fetch the manifest, its digest and container image configuration of the OCI object referenced by the given url.
     pub async fn manifest_and_config(
         &self,
@@ -259,11 +309,21 @@ impl Registry {
         let url: Url = Url::parse(format!("registry://{}", reference).as_str())?;
         let registry_auth = Registry::auth(reference.registry());
         let sources: Sources = sources.cloned().unwrap_or_default();
-        let cp = crate::client_protocol(&url, &sources)?;
 
-        let (manifest, digest, config) = Registry::client(cp)
-            .pull_manifest_and_config(&reference, &registry_auth)
-            .await?;
+        let (manifest, digest, config) = try_with_protocols(&url, &sources, |client_protocol| {
+            Box::pin({
+                let reference = reference.clone();
+                let registry_auth = registry_auth.clone();
+                async move {
+                    let res = Registry::client(client_protocol)
+                        .pull_manifest_and_config(&reference, &registry_auth)
+                        .await?;
+                    Ok(res)
+                }
+            })
+        })
+        .await?;
+
         let config_json = serde_json::from_str(&config)?;
 
         Ok((manifest, digest, config_json))

src/sources.rs

@@ -80,7 +80,7 @@ struct RawSources {
 struct RawCertificate(#[serde(with = "serde_bytes")] Vec<u8>);
 
 #[derive(Clone, Debug, Default)]
-struct SourceAuthorities(HashMap<String, Vec<Certificate>>);
+pub struct SourceAuthorities(pub HashMap<String, Vec<Certificate>>);
 
 impl TryFrom<RawSourceAuthorities> for SourceAuthorities {
     type Error = SourceError;
@@ -105,8 +105,8 @@ impl TryFrom<RawSourceAuthorities> for SourceAuthorities {
 
 #[derive(Clone, Debug, Default)]
 pub struct Sources {
-    insecure_sources: HashSet<String>,
-    source_authorities: SourceAuthorities,
+    pub insecure_sources: HashSet<String>,
+    pub source_authorities: SourceAuthorities,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]