diff --git a/airflow/hooks/S3_hook.py b/airflow/hooks/S3_hook.py
index 73b1154fb248b..ac259cd89ff1e 100644
--- a/airflow/hooks/S3_hook.py
+++ b/airflow/hooks/S3_hook.py
@@ -101,12 +101,13 @@ def __init__(
         self.extra_params = self.s3_conn.extra_dejson
         self.profile = self.extra_params.get('profile')
         self.calling_format = None
-        self._creds_in_conn = 'aws_secret_access_key' in self.extra_params
+        self._creds_in_conn = any(['aws_secret_access_key' in self.extra_params,
+                                   self.s3_conn.password])
         self._creds_in_config_file = 's3_config_file' in self.extra_params
         self._default_to_boto = False
         if self._creds_in_conn:
-            self._a_key = self.extra_params['aws_access_key_id']
-            self._s_key = self.extra_params['aws_secret_access_key']
+            self._a_key = self._get_access_key_id()
+            self._s_key = self._get_secret_access_key()
             if 'calling_format' in self.extra_params:
                 self.calling_format = self.extra_params['calling_format']
         elif self._creds_in_config_file:
@@ -135,6 +136,24 @@ def __setstate__(self, d):
         self.__dict__.update(d)
         self.__dict__['connection'] = self.get_conn()
 
+    def _get_access_key_id(self):
+        """
+        Access key ID can be provided in extras or as connection login.
+        """
+        access_key = self.extra_params.get('aws_access_key_id')
+        if access_key is not None:
+            return access_key
+        return self.s3_conn.login
+
+    def _get_secret_access_key(self):
+        """
+        Secret access key can be provided in extras or as connection password.
+        """
+        secret_key = self.extra_params.get('aws_secret_access_key')
+        if secret_key is not None:
+            return secret_key
+        return self.s3_conn.password
+
     def _parse_s3_url(self, s3url):
         warnings.warn(
             'Please note: S3Hook._parse_s3_url() is now '