v2: add content digest module and verify blob integrity on download

Cargo.toml

@@ -31,6 +31,7 @@ tar = "0.4"
 tokio = "0.1"
 dirs = "1.0"
 reqwest = { version = "^0.9.6", default-features = false }
+sha2 = "^0.8.0"
 
 [dev-dependencies]
 env_logger = "0.6"

examples/trace.rs

@@ -82,12 +82,7 @@ fn run(
                 .has_manifest(&image, &version, None)
                 .and_then(move |manifest_option| Ok((dclient, manifest_option)))
                 .and_then(|(dclient, manifest_option)| match manifest_option {
-                    None => {
-                        return Err(
-                            format!("{}:{} doesn't have a manifest", &image, &version).into()
-                        )
-                    }
-
+                    None => Err(format!("{}:{} doesn't have a manifest", &image, &version).into()),
                     Some(manifest_kind) => Ok((dclient, manifest_kind)),
                 })
         })

src/lib.rs

@@ -54,6 +54,7 @@ extern crate tar;
 #[macro_use]
 extern crate strum_macros;
 extern crate reqwest;
+extern crate sha2;
 
 pub mod errors;
 pub mod mediatypes;

src/v2/blobs.rs

@@ -36,24 +36,26 @@ impl Client {
 
     /// Retrieve blob.
     pub fn get_blob(&self, name: &str, digest: &str) -> FutureBlob {
-        let url = {
+        let fres_digest = futures::future::result(ContentDigest::try_new(digest.to_string()));
+
+        let fres_blob = {
             let ep = format!("{}/v2/{}/blobs/{}", self.base_url, name, digest);
-            match reqwest::Url::parse(&ep) {
-                Ok(url) => url,
-                Err(e) => {
-                    return Box::new(futures::future::err::<_, _>(Error::from(format!(
+            reqwest::Url::parse(&ep).map_err(|e|{
+                    ::errors::Error::from(format!(
                         "failed to parse url from string: {}",
                         e
-                    ))));
-                }
-            }
-        };
-
-        let fres = self.build_reqwest(reqwest::async::Client::new().get(url))
-            .send()
-            .map_err(|e| ::errors::Error::from(format!("{}", e)))
+                    ))
+            })
+            .map(|url|{
+                self.build_reqwest(reqwest::async::Client::new()
+                    .get(url))
+                    .send()
+                    .map_err(|e| ::errors::Error::from(format!("{}", e)))
+            })
+            .into_future()
+            .flatten()
             .and_then(|res| {
-                trace!("Blob GET status: {:?}", res.status());
+                trace!("GET {} status: {}", res.url(), res.status());
                 let status = res.status();
 
                 if status.is_success()
@@ -99,7 +101,14 @@ impl Client {
                         status
                     )))
                 }
-            });
+            })
+        };
+
+        let fres = fres_digest.join(fres_blob).and_then(|(digest, body)| {
+            digest.try_verify(&body)?;
+            Ok(body)
+        });
+
         Box::new(fres)
     }
 }

src/v2/content_digest.rs

@@ -0,0 +1,136 @@
+/// Implements types and methods for content verification
+use sha2::{self, Digest};
+use v2::*;
+
+/// ContentDigest stores a digest and its DigestAlgorithm
+#[derive(Clone, Debug, PartialEq)]
+pub(crate) struct ContentDigest {
+    digest: String,
+    algorithm: DigestAlgorithm,
+}
+
+/// DigestAlgorithm declares the supported algorithms
+#[derive(Display, Clone, Debug, PartialEq, EnumString)]
+enum DigestAlgorithm {
+    #[strum(to_string = "sha256")]
+    Sha256,
+}
+
+impl ContentDigest {
+    /// try_new attempts to parse the digest string and create a ContentDigest instance from it
+    ///
+    /// Success depends on
+    /// - the string having a "algorithm:" prefix
+    /// - the algorithm being supported by DigestAlgorithm
+    pub fn try_new(digest: String) -> Result<Self> {
+        let digest_split = digest.split(':').collect::<Vec<&str>>();
+
+        if digest_split.len() != 2 {
+            return Err(format!("digest '{}' does not have an algorithm prefix", digest).into());
+        }
+
+        let algorithm =
+            std::str::FromStr::from_str(digest_split[0]).map_err(|e| format!("{}", e))?;
+        Ok(ContentDigest {
+            digest: digest_split[1].to_string(),
+            algorithm,
+        })
+    }
+
+    /// try_verify hashes the input slice and compares it with the digest stored in this instance
+    ///
+    /// Success depends on the result of the comparison
+    pub fn try_verify(&self, input: &[u8]) -> Result<()> {
+        let hash = self.algorithm.hash(input);
+        let layer_digest = Self::try_new(hash)?;
+
+        if self != &layer_digest {
+            return Err(format!(
+                "content verification failed. expected '{}', got '{}'",
+                self.to_owned(),
+                layer_digest.to_owned()
+            )
+            .into());
+        }
+
+        trace!("content verification succeeded for '{}'", &layer_digest);
+        Ok(())
+    }
+}
+
+impl std::fmt::Display for ContentDigest {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "{}:{}", self.algorithm, self.digest)
+    }
+}
+
+impl DigestAlgorithm {
+    fn hash(&self, input: &[u8]) -> String {
+        match self {
+            DigestAlgorithm::Sha256 => {
+                let hash = sha2::Sha256::digest(input);
+                format!("{}:{:x}", self, hash)
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    type Fallible<T> = Result<T>;
+
+    #[test]
+    fn try_new_succeeds_with_correct_digest() -> Fallible<()> {
+        for correct_digest in
+            &["sha256:0000000000000000000000000000000000000000000000000000000000000000"]
+        {
+            ContentDigest::try_new(correct_digest.to_string())?;
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn try_new_succeeds_with_incorrect_digest() -> Fallible<()> {
+        for incorrect_digest in &[
+            "invalid",
+            "invalid:",
+            "invalid:0000000000000000000000000000000000000000000000000000000000000000",
+        ] {
+            if ContentDigest::try_new(incorrect_digest.to_string()).is_ok() {
+                return Err(format!(
+                    "expected try_new to fail for incorrect digest {}",
+                    incorrect_digest
+                )
+                .into());
+            }
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn try_verify_succeeds_with_same_content() -> Fallible<()> {
+        let blob: &[u8] = b"somecontent";
+        let digest = DigestAlgorithm::Sha256.hash(&blob);
+
+        ContentDigest::try_new(digest)?.try_verify(&blob)
+    }
+
+    #[test]
+    fn try_verify_fails_with_different_content() -> Fallible<()> {
+        let blob: &[u8] = b"somecontent";
+        let different_blob: &[u8] = b"someothercontent";
+        let digest = DigestAlgorithm::Sha256.hash(&blob);
+
+        if ContentDigest::try_new(digest)?
+            .try_verify(&different_blob)
+            .is_ok()
+        {
+            return Err("expected try_verify to fail for a different blob".into());
+        }
+
+        Ok(())
+    }
+}

src/v2/mod.rs

@@ -51,6 +51,9 @@ pub use self::tags::StreamTags;
 mod blobs;
 pub use self::blobs::FutureBlob;
 
+mod content_digest;
+pub(crate) use self::content_digest::ContentDigest;
+
 /// A Client to make outgoing API requests to a registry.
 #[derive(Clone, Debug)]
 pub struct Client {

tests/mock/blobs_download.rs

@@ -1,9 +1,13 @@
 extern crate dkregistry;
 extern crate mockito;
+extern crate sha2;
 extern crate tokio;
 
 use self::mockito::mock;
 use self::tokio::runtime::current_thread::Runtime;
+use mock::{self, blobs_download::sha2::Digest};
+
+type Fallible<T> = Result<T, Box<std::error::Error>>;
 
 #[test]
 fn test_blobs_has_layer() {
@@ -61,3 +65,69 @@ fn test_blobs_hasnot_layer() {
 
     mockito::reset();
 }
+
+#[test]
+fn get_blobs_succeeds_with_consistent_layer() -> Fallible<()> {
+    let addr = mockito::server_address().to_string();
+
+    let name = "my-repo/my-image";
+    let blob = b"hello";
+    let digest = format!("sha256:{:x}", sha2::Sha256::digest(blob));
+
+    let ep = format!("/v2/{}/blobs/{}", &name, &digest);
+    let _m = mock("GET", ep.as_str())
+        .with_status(200)
+        .with_body(blob)
+        .create();
+
+    let mut runtime = Runtime::new().unwrap();
+    let dclient = dkregistry::v2::Client::configure()
+        .registry(&addr)
+        .insecure_registry(true)
+        .username(None)
+        .password(None)
+        .build()
+        .unwrap();
+
+    let futcheck = dclient.get_blob(&name, &digest);
+
+    let result = runtime.block_on(futcheck)?;
+    assert_eq!(blob, result.as_slice());
+
+    mockito::reset();
+    Ok(())
+}
+
+#[test]
+fn get_blobs_fails_with_inconsistent_layer() -> Fallible<()> {
+    let addr = mockito::server_address().to_string();
+
+    let name = "my-repo/my-image";
+    let blob = b"hello";
+    let blob2 = b"hello2";
+    let digest = format!("sha256:{:x}", sha2::Sha256::digest(blob));
+
+    let ep = format!("/v2/{}/blobs/{}", &name, &digest);
+    let _m = mock("GET", ep.as_str())
+        .with_status(200)
+        .with_body(blob2)
+        .create();
+
+    let mut runtime = Runtime::new().unwrap();
+    let dclient = dkregistry::v2::Client::configure()
+        .registry(&addr)
+        .insecure_registry(true)
+        .username(None)
+        .password(None)
+        .build()
+        .unwrap();
+
+    let futcheck = dclient.get_blob(&name, &digest);
+
+    if runtime.block_on(futcheck).is_ok() {
+        return Err("expected get_blob to fail with an inconsistent blob".into());
+    };
+
+    mockito::reset();
+    Ok(())
+}