[WIP] v2/blobs: verify 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/image.rs

@@ -117,7 +117,7 @@ fn run(
             )
         })
         .and_then(|(dclient, layers)| {
-            let image = image.clone();
+            let image = image.to_owned();
 
             println!("{} -> got {} layer(s)", &image, layers.len(),);
 

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,6 +36,8 @@ impl Client {
 
     /// Retrieve blob.
     pub fn get_blob(&self, name: &str, digest: &str) -> FutureBlob {
+        // let layer_digest = ContentDigest::try_new(digest.to_string());
+
         let url = {
             let ep = format!("{}/v2/{}/blobs/{}", self.base_url, name, digest);
             match reqwest::Url::parse(&ep) {
@@ -49,11 +51,13 @@ impl Client {
             }
         };
 
-        let fres = self.build_reqwest(reqwest::async::Client::new().get(url))
+        let fres_digest = futures::future::result(ContentDigest::try_new(digest.to_string()));
+
+        let fres_reqwest = self.build_reqwest(reqwest::async::Client::new().get(url))
             .send()
             .map_err(|e| ::errors::Error::from(format!("{}", e)))
             .and_then(|res| {
-                trace!("Blob GET status: {:?}", res.status());
+                trace!("GET {} status: {}", res.url(), res.status());
                 let status = res.status();
 
                 if status.is_success()
@@ -100,6 +104,12 @@ impl Client {
                     )))
                 }
             });
+
+        let fres = fres_digest.join(fres_reqwest).and_then(|(digest, body)| {
+            digest.try_verify(&body)?;
+            Ok(body)
+        });
+
         Box::new(fres)
     }
 }

src/v2/content_digest.rs

@@ -0,0 +1,92 @@
+/// 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 struct ContentDigest {
+    digest: String,
+    algorithm: DigestAlgorithm,
+}
+
+/// DigestAlgorithm declares the supported algorithms
+#[derive(Display, Clone, Debug, PartialEq, EnumString)]
+pub enum DigestAlgorithm {
+    sha256,
+}
+
+impl ContentDigest {
+    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..].join(""),
+            algorithm,
+        })
+    }
+
+    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::*;
+
+    #[test]
+    fn try_verify_succeeds_with_same_content() -> Result<()> {
+        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() -> Result<()> {
+        let blob: &[u8] = b"somecontent";
+        let different_blob: &[u8] = b"someothercontent";
+        let digest = DigestAlgorithm::sha256.hash(&blob);
+
+        match ContentDigest::try_new(digest)?.try_verify(&different_blob) {
+            Ok(()) => Err("expected try_verify to fail for a different blob".into()),
+            Err(_) => Ok(()),
+        }
+    }
+}

src/v2/mod.rs

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