storage: download integrity check

lib/storage/file.js

@@ -211,6 +211,19 @@ File.prototype.copy = function(destination, callback) {
  * piped to a writable stream or listened to for 'data' events to read a file's
  * contents.
  *
+ * In the unlikely event there is a mismatch between what you downloaded and the
+ * version in your Bucket, your error handler will receive an error with code
+ * "CONTENT_DOWNLOAD_MISMATCH". If you receive this error, the best recourse is
+ * to try downloading the file again.
+ *
+ * @param {object=} options - Configuration object.
+ * @param {string|boolean} options.validation - Possible values: `"md5"`,
+ * `"crc32c"`, or `false`. By default, data integrity is validated with an
+ * MD5 checksum for maximum reliability, falling back to CRC32c when an MD5
+ * hash wasn't returned from the API. CRC32c will provide better performance
+ * with less reliability. You may also choose to skip validation completely,
+ * however this is **not recommended**.
+ *
  * @example
  * //-
  * // <h4>Downloading a File</h4>
@@ -226,35 +239,133 @@ File.prototype.copy = function(destination, callback) {
  * .pipe(fs.createWriteStream('/Users/stephen/Photos/image.png'))
  * .on('error', function(err) {});
  */
-File.prototype.createReadStream = function() {
- var storage = this.bucket.storage;
- var dup = duplexify();
- function createAuthorizedReq(uri) {
- var reqOpts = { uri: uri };
- storage.makeAuthorizedRequest_(reqOpts, {
- onAuthorized: function(err, authorizedReqOpts) {
- if (err) {
- dup.emit('error', err);
- dup.end();
- return;
- }
- dup.setReadable(request(authorizedReqOpts));
- }
- });
+File.prototype.createReadStream = function(options) {
+ options = options || {};
+
+ var that = this;
+ var throughStream = through();
+
+ var validations = ['crc32c', 'md5'];
+ var validation;
+
+ if (util.is(options.validation, 'string')) {
+ options.validation = options.validation.toLowerCase();
+
+ if (validations.indexOf(options.validation) > -1) {
+ validation = options.validation;
+ } else {
+ validation = 'all';
+ }
+ }
+
+ if (util.is(options.validation, 'undefined')) {
+ validation = 'all';
  }
+
+ var crc32c = validation === 'crc32c' || validation === 'all';
+ var md5 = validation === 'md5' || validation === 'all';
+
  if (this.metadata.mediaLink) {
  createAuthorizedReq(this.metadata.mediaLink);
  } else {
  this.getMetadata(function(err, metadata) {
  if (err) {
- dup.emit('error', err);
- dup.end();
+ throughStream.emit('error', err);
+ throughStream.end();
  return;
  }
+
  createAuthorizedReq(metadata.mediaLink);
  });
  }
- return dup;
+
+ return throughStream;
+
+ // Authenticate the request, then pipe the remote API request to the stream
+ // returned to the user.
+ function createAuthorizedReq(uri) {
+ var reqOpts = {
+ uri: uri
+ };
+
+ that.bucket.storage.makeAuthorizedRequest_(reqOpts, {
+ onAuthorized: function(err, authorizedReqOpts) {
+ if (err) {
+ throughStream.emit('error', err);
+ throughStream.end();
+ return;
+ }
+
+ // For data integrity, hash the contents of the stream as we receive it
+ // from the server.
+ var localCrc32cHash;
+ var localMd5Hash = crypto.createHash('md5');
+
+ request(authorizedReqOpts)
+ .on('error', function(err) {
+ throughStream.emit('error', err);
+ throughStream.end();
+ })
+
+ .on('data', function(chunk) {
+ if (crc32c) {
+ localCrc32cHash = crc.calculate(chunk, localCrc32cHash);
+ }
+
+ if (md5) {
+ localMd5Hash.update(chunk);
+ }
+ })
+
+ .on('complete', function(res) {
+ var failed = false;
+ var crcFail = true;
+ var md5Fail = true;
+
+ var hashes = {};
+ res.headers['x-goog-hash'].split(',').forEach(function(hash) {
+ var hashType = hash.split('=')[0];
+ hashes[hashType] = hash.substr(hash.indexOf('=') + 1);
+ });
+
+ var remoteMd5 = hashes.md5;
+ var remoteCrc = hashes.crc32c && hashes.crc32c.substr(4);
+
+ if (crc32c) {
+ crcFail =
+ new Buffer([localCrc32cHash]).toString('base64') !== remoteCrc;
+ failed = crcFail;
+ }
+
+ if (md5) {
+ md5Fail = localMd5Hash.digest('base64') !== remoteMd5;
+ failed = md5Fail;
+ }
+
+ if (validation === 'all') {
+ failed = remoteMd5 ? md5Fail : crcFail;
+ }
+
+ if (failed) {
+ var error = new Error([
+ 'The downloaded data did not match the data from the server.',
+ 'To be sure the content is the same, you should download the',
+ 'file again.'
+ ].join(' '));
+ error.code = 'CONTENT_DOWNLOAD_MISMATCH';
+
+ throughStream.emit('error', error);
+ } else {
+ throughStream.emit('complete');
+ }
+
+ throughStream.end();
+ })
+
+ .pipe(throughStream);
+ }
+ });
+ }
 };
 
 /**
@@ -688,7 +799,7 @@ File.prototype.startResumableUpload_ = function(stream, metadata) {
  method: 'PUT',
  uri: resumableUri
  }, {
- onAuthorized: function (err, reqOpts) {
+ onAuthorized: function(err, reqOpts) {
  if (err) {
  handleError(err);
  return;

regression/storage.js

@@ -322,12 +322,17 @@ describe('storage', function() {
 
  writeStream.on('error', done);
  writeStream.on('complete', function() {
+ var data = new Buffer('');
+
  file.createReadStream()
+ .on('error', done)
  .on('data', function(chunk) {
- assert.equal(String(chunk), contents);
+ data = Buffer.concat([data, chunk]);
  })
- .on('error', done)
- .on('end', done);
+ .on('complete', function() {
+ assert.equal(data.toString(), contents);
+ done();
+ });
  });
  });
 

test/storage/file.js

@@ -263,13 +263,15 @@ describe('File', function() {
  });
 
  it('should create an authorized request', function(done) {
- request_Override = function(opts) {
+ file.bucket.storage.makeAuthorizedRequest_ = function(opts) {
  assert.equal(opts.uri, metadata.mediaLink);
  done();
  };
+
  file.getMetadata = function(callback) {
  callback(null, metadata);
  };
+
  file.createReadStream();
  });
 
@@ -292,33 +294,129 @@ describe('File', function() {
 
  it('should get readable stream from request', function(done) {
  var fakeRequest = { a: 'b', c: 'd' };
- file.getMetadata = function(callback) {
- callback(null, metadata);
- };
+
+ // Faking a stream implementation so we can simulate an actual Request
+ // request. The only thing we want to know is if the data passed to
+ // request was correct.
  request_Override = function(req) {
+ if (!(this instanceof request_Override)) {
+ return new request_Override(req);
+ }
+
+ stream.Readable.call(this);
+ this._read = util.noop;
+
  assert.deepEqual(req, fakeRequest);
  done();
  };
- file.bucket.storage.makeAuthorizedRequest_ = function(opts, callback) {
- (callback.onAuthorized || callback)(null, fakeRequest);
- };
- file.createReadStream();
- });
+ nodeutil.inherits(request_Override, stream.Readable);
 
- it('should set readable stream', function() {
- var dup = duplexify();
  file.getMetadata = function(callback) {
  callback(null, metadata);
  };
- request_Override = function() {
- return dup;
- };
+
  file.bucket.storage.makeAuthorizedRequest_ = function(opts, callback) {
- (callback.onAuthorized || callback)();
+ (callback.onAuthorized || callback)(null, fakeRequest);
  };
+
  file.createReadStream();
- assert.deepEqual(readableStream, dup);
- readableStream = null;
+ });
+
+ describe('validation', function() {
+ var data = 'test';
+
+ var crc32cBase64 = new Buffer([crc.calculate(data)]).toString('base64');
+
+ var md5HashBase64 = crypto.createHash('md5');
+ md5HashBase64.update(data);
+ md5HashBase64 = md5HashBase64.digest('base64');
+
+ var fakeResponse = {
+ crc32c: {
+ headers: { 'x-goog-hash': 'crc32c=####' + crc32cBase64 }
+ },
+ md5: {
+ headers: { 'x-goog-hash': 'md5=' + md5HashBase64 }
+ }
+ };
+
+ function getFakeRequest(data, fakeResponse) {
+ function FakeRequest(req) {
+ if (!(this instanceof FakeRequest)) {
+ return new FakeRequest(req);
+ }
+
+ var that = this;
+
+ stream.Readable.call(this);
+ this._read = function() {
+ this.push(data);
+ this.push(null);
+ };
+
+ setImmediate(function() {
+ that.emit('complete', fakeResponse);
+ });
+ }
+ nodeutil.inherits(FakeRequest, stream.Readable);
+ return FakeRequest;
+ }
+
+ beforeEach(function() {
+ file.metadata.mediaLink = 'http://uri';
+
+ file.bucket.storage.makeAuthorizedRequest_ = function(opts, callback) {
+ (callback.onAuthorized || callback)(null, {});
+ };
+ });
+
+ it('should validate with crc32c', function(done) {
+ request_Override = getFakeRequest(data, fakeResponse.crc32c);
+
+ file.createReadStream({ validation: 'crc32c' })
+ .on('error', done)
+ .on('complete', done);
+ });
+
+ it('should emit an error if crc32c validation fails', function(done) {
+ request_Override = getFakeRequest('bad-data', fakeResponse.crc32c);
+
+ file.createReadStream({ validation: 'crc32c' })
+ .on('error', function(err) {
+ assert.equal(err.code, 'CONTENT_DOWNLOAD_MISMATCH');
+ done();
+ });
+ });
+
+ it('should validate with md5', function(done) {
+ request_Override = getFakeRequest(data, fakeResponse.md5);
+
+ file.createReadStream({ validation: 'md5' })
+ .on('error', done)
+ .on('complete', done);
+ });
+
+ it('should emit an error if md5 validation fails', function(done) {
+ request_Override = getFakeRequest('bad-data', fakeResponse.crc32c);
+
+ file.createReadStream({ validation: 'md5' })
+ .on('error', function(err) {
+ assert.equal(err.code, 'CONTENT_DOWNLOAD_MISMATCH');
+ done();
+ });
+ });
+
+ it('should default to md5 validation', function(done) {
+ request_Override = getFakeRequest(data, {
+ headers: { 'x-goog-hash': 'md5=fakefakefake' }
+ });
+
+ file.createReadStream()
+ .on('error', function(err) {
+ assert.equal(err.code, 'CONTENT_DOWNLOAD_MISMATCH');
+ done();
+ });
+ });
  });
  });