ssh: check if cache exists when saving external outputs
Right now, we are not checking if the file already exists when saving it, this would solve some issues regarding file permissions
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
index 6dda03602..9952ce824 100644
--- a/dvc/remote/base.py
+++ b/dvc/remote/base.py
@@ -173,7 +173,21 @@ class RemoteBase(object):
         return False
 
     def save(self, path_info, checksum_info):
-        raise RemoteActionNotImplemented("save", self.scheme)
+        if path_info["scheme"] != self.scheme:
+            raise RemoteActionNotImplemented(
+                "save {} -> {}".format(path_info["scheme"], self.scheme),
+                self.scheme,
+            )
+
+        checksum = checksum_info[self.PARAM_CHECKSUM]
+        if not self.changed_cache(checksum):
+            return
+
+        to_info = self.checksum_to_path_info(checksum)
+
+        logger.info("Saving '{}' to '{}'.".format(path_info, to_info))
+
+        self.copy(path_info, to_info)
 
     def download(
         self,
@@ -265,6 +279,10 @@ class RemoteBase(object):
     def changed_cache(self, checksum):
         cache = self.checksum_to_path_info(checksum)
         expected = {self.PARAM_CHECKSUM: checksum}
+
+        if not self.exists(cache):
+            return True
+
         actual = self.save_info(cache)
 
         logger.debug(
diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py
index df9e5af7c..aa7d8ffd6 100644
--- a/dvc/remote/gs.py
+++ b/dvc/remote/gs.py
@@ -87,18 +87,6 @@ class RemoteGS(RemoteBase):
             blob, self.gs.bucket(to_info["bucket"]), new_name=to_info["path"]
         )
 
-    def save(self, path_info):
-        if path_info["scheme"] != "gs":
-            raise NotImplementedError
-
-        md5 = self.get_md5(path_info["bucket"], path_info["path"])
-        path = self.checksum_to_path(md5)
-        to_info = {"scheme": "gs", "bucket": self.bucket, "path": path}
-
-        self.copy(path_info, to_info)
-
-        return {self.PARAM_CHECKSUM: md5}
-
     def remove(self, path_info):
         if path_info["scheme"] != "gs":
             raise NotImplementedError
diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py
index 287bb66ae..337b777b1 100644
--- a/dvc/remote/hdfs.py
+++ b/dvc/remote/hdfs.py
@@ -88,18 +88,6 @@ class RemoteHDFS(RemoteBase):
 
         return {self.PARAM_CHECKSUM: self.checksum(path_info)}
 
-    def save(self, path_info, checksum_info):
-        if path_info["scheme"] != "hdfs":
-            raise NotImplementedError
-
-        assert path_info.get("path")
-
-        checksum = checksum_info[self.PARAM_CHECKSUM]
-        dest = path_info.copy()
-        dest["path"] = self.checksum_to_path(checksum)
-
-        self.copy(path_info, dest)
-
     def remove(self, path_info):
         if path_info["scheme"] != "hdfs":
             raise NotImplementedError
diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py
index 44186b52c..e38369d52 100644
--- a/dvc/remote/s3.py
+++ b/dvc/remote/s3.py
@@ -109,16 +109,6 @@ class RemoteS3(RemoteBase):
         source = {"Bucket": from_info["bucket"], "Key": from_info["path"]}
         self.s3.copy(source, to_info["bucket"], to_info["path"])
 
-    def save(self, path_info, checksum_info):
-        if path_info["scheme"] != "s3":
-            raise NotImplementedError
-
-        etag = checksum_info[self.PARAM_CHECKSUM]
-        path = self.checksum_to_path(etag)
-        to_info = {"scheme": "s3", "bucket": self.bucket, "path": path}
-
-        self.copy(path_info, to_info)
-
     def remove(self, path_info):
         if path_info["scheme"] != "s3":
             raise NotImplementedError
diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py
index 5340d7e6b..862247a56 100644
--- a/dvc/remote/ssh/__init__.py
+++ b/dvc/remote/ssh/__init__.py
@@ -115,16 +115,6 @@ class RemoteSSH(RemoteBase):
 
         return {self.PARAM_CHECKSUM: self.md5(path_info)}
 
-    def save(self, path_info, checksum_info):
-        if path_info["scheme"] != self.scheme:
-            raise NotImplementedError
-
-        md5 = checksum_info[self.PARAM_CHECKSUM]
-        dest = path_info.copy()
-        dest["path"] = self.checksum_to_path(md5)
-
-        self.copy(path_info, dest)
-
     def remove(self, path_info):
         if path_info["scheme"] != self.scheme:
             raise NotImplementedError
diff --git a/tests/unit/remote/base.py b/tests/unit/remote/base.py
index c21b8105d..95216cfad 100644
--- a/tests/unit/remote/base.py
+++ b/tests/unit/remote/base.py
@@ -33,3 +33,29 @@ class TestCmdError(TestCase):
             ):
                 with self.assertRaises(RemoteCmdError):
                     remote_class(repo, config).remove("file")
+
+
+class TestCacheExists(TestCase):
+    def test(self):
+        for remote_class in REMOTES:
+            config = {
+                "url": remote_class.scheme + "://example/prefix",
+                "connection_string": "1234567",
+            }
+            remote = remote_class(None, config)
+
+            with mock.patch.object(remote, "changed_cache", return_value=True):
+                with mock.patch.object(remote, "copy") as cp:
+                    remote.save(
+                        remote.path_info, {remote.PARAM_CHECKSUM: "1234567890"}
+                    )
+                    cp.assert_called_once()
+
+            with mock.patch.object(
+                remote, "changed_cache", return_value=False
+            ):
+                with mock.patch.object(remote, "copy") as cp:
+                    remote.save(
+                        remote.path_info, {remote.PARAM_CHECKSUM: "1234567890"}
+                    )
+                    cp.assert_not_called()
['dvc/remote/hdfs.py', 'dvc/remote/gs.py', 'dvc/remote/s3.py', 'tests/unit/remote/base.py', 'dvc/remote/ssh/__init__.py', 'dvc/remote/base.py']