CredentialsProvider class added to support password rotation

CHANGES

@@ -23,6 +23,7 @@
     * ClusterPipeline Doesn't Handle ConnectionError for Dead Hosts (#2225)
     * Remove compatibility code for old versions of Hiredis, drop Packaging dependency
     * The `deprecated` library is no longer a dependency
+    * Added CredentialsProvider class to support password rotation
 
 * 4.1.3 (Feb 8, 2022)
     * Fix flushdb and flushall (#1926)

docs/examples/connection_examples.ipynb

@@ -97,6 +97,197 @@
     "user_connection.ping()"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "source": [
+    "## Connecting to a redis instance with username and password credential provider"
+   ],
+   "metadata": {}
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "outputs": [],
+   "source": [
+    "import redis\n",
+    "\n",
+    "creds_provider = redis.UsernamePasswordCredentialProvider(\"username\", \"password\")\n",
+    "user_connection = redis.Redis(host=\"localhost\", port=6379, credential_provider=creds_provider)\n",
+    "user_connection.ping()"
+   ],
+   "metadata": {}
+   }
+  },
+  {
+   "cell_type": "markdown",
+   "source": [
+    "## Connecting to a redis instance with standard credential provider"
+   ],
+   "metadata": {}
+   }
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "outputs": [],
+   "source": [
+    "from typing import Tuple\n",
+    "import redis\n",
+    "\n",
+    "creds_map = {\"user_1\": \"pass_1\",\n",
+    "             \"user_2\": \"pass_2\"}\n",
+    "\n",
+    "class UserMapCredentialProvider(redis.CredentialProvider):\n",
+    "    def __init__(self, username: str):\n",
+    "        self.username = username\n",
+    "\n",
+    "    def get_credentials(self) -> Tuple[str, str]:\n",
+    "        return self.username, creds_map.get(self.username)\n",
+    "\n",
+    "# Create a default connection to set the ACL user\n",
+    "default_connection = redis.Redis(host=\"localhost\", port=6379)\n",
+    "default_connection.acl_setuser(\n",
+    "    \"user_1\",\n",
+    "    enabled=True,\n",
+    "    passwords=[\"+\" + \"pass_1\"],\n",
+    "    keys=\"~*\",\n",
+    "    commands=[\"+ping\", \"+command\", \"+info\", \"+select\", \"+flushdb\"],\n",
+    ")\n",
+    "\n",
+    "# Create a UserMapCredentialProvider instance for user_1\n",
+    "creds_provider = UserMapCredentialProvider(\"user_1\")\n",
+    "# Initiate user connection with the credential provider\n",
+    "user_connection = redis.Redis(host=\"localhost\", port=6379,\n",
+    "                              credential_provider=creds_provider)\n",
+    "user_connection.ping()"
+   ],
+   "metadata": {}
+   }
+  },
+  {
+   "cell_type": "markdown",
+   "source": [
+    "## Connecting to a redis instance first with an initial credential set and then calling the credential provider"
+   ],
+   "metadata": {}
+   }
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "outputs": [],
+   "source": [
+    "from typing import Union\n",
+    "import redis\n",
+    "\n",
+    "class InitCredsSetCredentialProvider(redis.CredentialProvider):\n",
+    "    def __init__(self, username, password):\n",
+    "        self.username = username\n",
+    "        self.password = password\n",
+    "        self.call_supplier = False\n",
+    "\n",
+    "    def call_external_supplier(self) -> Union[Tuple[str], Tuple[str, str]]:\n",
+    "        # Call to an external credential supplier\n",
+    "        raise NotImplementedError\n",
+    "\n",
+    "    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:\n",
+    "        if self.call_supplier:\n",
+    "            return self.call_external_supplier()\n",
+    "        # Use the init set only for the first time\n",
+    "        self.call_supplier = True\n",
+    "        return self.username, self.password\n",
+    "\n",
+    "cred_provider = InitCredsSetCredentialProvider(username=\"init_user\", password=\"init_pass\")"
+   ],
+   "metadata": {}
+   }
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {
+    "collapsed": false
+   },
+   "source": [
+    "## Connecting to a redis instance with AWS Secrets Manager credential provider."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "import redis\n",
+    "import boto3\n",
+    "import json\n",
+    "import cachetools.func\n",
+    "\n",
+    "sm_client = boto3.client('secretsmanager')\n",
+    "  \n",
+    "def sm_auth_provider(self, secret_id, version_id=None, version_stage='AWSCURRENT'):\n",
+    "    @cachetools.func.ttl_cache(maxsize=128, ttl=24 * 60 * 60) #24h\n",
+    "    def get_sm_user_credentials(secret_id, version_id, version_stage):\n",
+    "        secret = sm_client.get_secret_value(secret_id, version_id)\n",
+    "        return json.loads(secret['SecretString'])\n",
+    "    creds = get_sm_user_credentials(secret_id, version_id, version_stage)\n",
+    "    return creds['username'], creds['password']\n",
+    "\n",
+    "secret_id = \"EXAMPLE1-90ab-cdef-fedc-ba987SECRET1\"\n",
+    "creds_provider = redis.CredentialProvider(supplier=sm_auth_provider, secret_id=secret_id)\n",
+    "user_connection = redis.Redis(host=\"localhost\", port=6379, credential_provider=creds_provider)\n",
+    "user_connection.ping()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Connecting to a redis instance with ElastiCache IAM credential provider."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 4,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "True"
+      ]
+     },
+     "execution_count": 4,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "import redis\n",
+    "import boto3\n",
+    "import cachetools.func\n",
+    "\n",
+    "ec_client = boto3.client('elasticache')\n",
+    "\n",
+    "def iam_auth_provider(self, user, endpoint, port=6379, region=\"us-east-1\"):\n",
+    "    @cachetools.func.ttl_cache(maxsize=128, ttl=15 * 60) # 15m\n",
+    "    def get_iam_auth_token(user, endpoint, port, region):\n",
+    "        return ec_client.generate_iam_auth_token(user, endpoint, port, region)\n",
+    "    iam_auth_token = get_iam_auth_token(endpoint, port, user, region)\n",
+    "    return iam_auth_token\n",
+    "\n",
+    "username = \"barshaul\"\n",
+    "endpoint = \"test-001.use1.cache.amazonaws.com\"\n",
+    "creds_provider = redis.CredentialProvider(supplier=iam_auth_provider, user=username,\n",
+    "                                           endpoint=endpoint)\n",
+    "user_connection = redis.Redis(host=endpoint, port=6379, credential_provider=creds_provider)\n",
+    "user_connection.ping()"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},
@@ -176,4 +367,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 2
-}
+}

redis/__init__.py

@@ -9,6 +9,7 @@
     SSLConnection,
     UnixDomainSocketConnection,
 )
+from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
 from redis.exceptions import (
     AuthenticationError,
     AuthenticationWrongNumberOfArgsError,
@@ -62,6 +63,7 @@ def int_or_str(value):
     "Connection",
     "ConnectionError",
     "ConnectionPool",
+    "CredentialProvider",
     "DataError",
     "from_url",
     "InvalidResponse",
@@ -76,6 +78,7 @@ def int_or_str(value):
     "SentinelManagedConnection",
     "SentinelManagedSSLConnection",
     "SSLConnection",
+    "UsernamePasswordCredentialProvider",
     "StrictRedis",
     "TimeoutError",
     "UnixDomainSocketConnection",

redis/client.py

@@ -5,6 +5,7 @@
 import time
 import warnings
 from itertools import chain
+from typing import Optional
 
 from redis.commands import (
     CoreCommands,
@@ -13,6 +14,7 @@
     list_or_args,
 )
 from redis.connection import ConnectionPool, SSLConnection, UnixDomainSocketConnection
+from redis.credentials import CredentialProvider
 from redis.exceptions import (
     ConnectionError,
     ExecAbortError,
@@ -938,6 +940,7 @@ def __init__(
         username=None,
         retry=None,
         redis_connect_func=None,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
         """
         Initialize a new Redis client.
@@ -985,6 +988,7 @@ def __init__(
                 "health_check_interval": health_check_interval,
                 "client_name": client_name,
                 "redis_connect_func": redis_connect_func,
+                "credential_provider": credential_provider,
             }
             # based on input, setup appropriate connection args
             if unix_socket_path is not None:

redis/cluster.py

@@ -121,6 +121,7 @@ def parse_cluster_shards(resp, **options):
     "connection_class",
     "connection_pool",
     "client_name",
+    "credential_provider",
     "db",
     "decode_responses",
     "encoding",

redis/connection.py

@@ -8,9 +8,11 @@
 from itertools import chain
 from queue import Empty, Full, LifoQueue
 from time import time
+from typing import Optional
 from urllib.parse import parse_qs, unquote, urlparse
 
 from redis.backoff import NoBackoff
+from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
 from redis.exceptions import (
     AuthenticationError,
     AuthenticationWrongNumberOfArgsError,
@@ -502,6 +504,7 @@ def __init__(
         username=None,
         retry=None,
         redis_connect_func=None,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
         """
         Initialize a new Connection.
@@ -514,9 +517,18 @@ def __init__(
         self.host = host
         self.port = int(port)
         self.db = db
-        self.username = username
         self.client_name = client_name
+        if (username or password) and credential_provider is not None:
+            raise DataError(
+                "'username' and 'password' cannot be passed along with 'credential_"
+                "provider'. Please provide only one of the following arguments: \n"
+                "1. 'password' and (optional) 'username'\n"
+                "2. 'credential_provider'"
+            )
+
+        self.credential_provider = credential_provider
         self.password = password
+        self.username = username
         self.socket_timeout = socket_timeout
         self.socket_connect_timeout = socket_connect_timeout or socket_timeout
         self.socket_keepalive = socket_keepalive
@@ -675,12 +687,13 @@ def on_connect(self):
         "Initialize the connection, authenticate and select a database"
         self._parser.on_connect(self)
 
-        # if username and/or password are set, authenticate
-        if self.username or self.password:
-            if self.username:
-                auth_args = (self.username, self.password or "")
-            else:
-                auth_args = (self.password,)
+        # if credential provider or username and/or password are set, authenticate
+        if self.credential_provider or (self.username or self.password):
+            cred_provider = (
+                self.credential_provider
+                or UsernamePasswordCredentialProvider(self.username, self.password)
+            )
+            auth_args = cred_provider.get_credentials()
             # avoid checking health here -- PING will fail if we try
             # to check the health prior to the AUTH
             self.send_command("AUTH", *auth_args, check_health=False)
@@ -692,7 +705,7 @@ def on_connect(self):
                 # server seems to be < 6.0.0 which expects a single password
                 # arg. retry auth with just the password.
                 # https://github.com/andymccurdy/redis-py/issues/1274
-                self.send_command("AUTH", self.password, check_health=False)
+                self.send_command("AUTH", auth_args[-1], check_health=False)
                 auth_response = self.read_response()
 
             if str_if_bytes(auth_response) != "OK":
@@ -1050,6 +1063,7 @@ def __init__(
         client_name=None,
         retry=None,
         redis_connect_func=None,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
         """
         Initialize a new UnixDomainSocketConnection.
@@ -1061,9 +1075,17 @@ def __init__(
         self.pid = os.getpid()
         self.path = path
         self.db = db
-        self.username = username
         self.client_name = client_name
+        if (username or password) and credential_provider is not None:
+            raise DataError(
+                "'username' and 'password' cannot be passed along with 'credential_"
+                "provider'. Please provide only one of the following arguments: \n"
+                "1. 'password' and (optional) 'username'\n"
+                "2. 'credential_provider'"
+            )
+        self.credential_provider = credential_provider
         self.password = password
+        self.username = username
         self.socket_timeout = socket_timeout
         self.retry_on_timeout = retry_on_timeout
         if retry_on_error is SENTINEL:

redis/credentials.py

@@ -0,0 +1,26 @@
+from typing import Optional, Tuple, Union
+
+
+class CredentialProvider:
+    """
+    Credentials Provider.
+    """
+
+    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:
+        raise NotImplementedError("get_credentials must be implemented")
+
+
+class UsernamePasswordCredentialProvider(CredentialProvider):
+    """
+    Simple implementation of CredentialProvider that just wraps static
+    username and password.
+    """
+
+    def __init__(self, username: Optional[str] = None, password: Optional[str] = None):
+        self.username = username or ""
+        self.password = password or ""
+
+    def get_credentials(self):
+        if self.username:
+            return self.username, self.password
+        return (self.password,)