CredentialsProvider class added to support password rotation (#2261)

* A CredentialsProvider class has been added to allow the user to add his own provider for password rotation

* Moved CredentialsProvider to a separate file, added type hints

* Changed username and password to properties

* Added: StaticCredentialProvider, examples, tests
Changed: CredentialsProvider to CredentialProvider
Fixed: calling AUTH only with password

* Changed private members' prefix to __

* fixed linters

* fixed auth test

* fixed credential test

* Raise an error if username or password are passed along with credential_provider

* fixing linters

* fixing test

* Changed dundered to single per side underscore

* Changed Connection class members username and password to properties to enable backward compatibility with changing the members value on existing connection.

* Reverting last commit and adding backward compatibility to 'username' and 'password' inside on_connect function

* Refactored CredentialProvider class

* Fixing tuple type to Tuple

* Fixing optional string members in UsernamePasswordCredentialProvider

* Fixed credential test

* Added credential provider support to AsyncRedis

* linters

* linters

* linters

* linters - black

Co-authored-by: dvora-h <67596500+dvora-h@users.noreply.github.com>
Co-authored-by: dvora-h <dvora.heller@redis.com>

2 files changed, 268 insertions, 54 deletions

diff --git a/docs/examples/asyncio_examples.ipynb b/docs/examples/asyncio_examples.ipynb
index dab7a96..855255c 100644
--- a/docs/examples/asyncio_examples.ipynb
+++ b/docs/examples/asyncio_examples.ipynb
@@ -21,11 +21,6 @@
   {
    "cell_type": "code",
    "execution_count": 1,
-   "metadata": {
-    "pycharm": {
-     "name": "#%%\n"
-    }
-   },
    "outputs": [
     {
      "name": "stdout",
@@ -41,27 +36,29 @@
     "connection = redis.Redis()\n",
     "print(f\"Ping successful: {await connection.ping()}\")\n",
     "await connection.close()"
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
   },
   {
    "cell_type": "markdown",
+   "source": [
+    "If you supply a custom `ConnectionPool` that is supplied to several `Redis` instances, you may want to disconnect the connection pool explicitly. Disconnecting the connection pool simply disconnects all connections hosted in the pool."
+   ],
    "metadata": {
+    "collapsed": false,
     "pycharm": {
      "name": "#%% md\n"
     }
-   },
-   "source": [
-    "If you supply a custom `ConnectionPool` that is supplied to several `Redis` instances, you may want to disconnect the connection pool explicitly. Disconnecting the connection pool simply disconnects all connections hosted in the pool."
-   ]
+   }
   },
   {
    "cell_type": "code",
    "execution_count": 2,
-   "metadata": {
-    "pycharm": {
-     "name": "#%%\n"
-    }
-   },
    "outputs": [],
    "source": [
     "import redis.asyncio as redis\n",
@@ -70,15 +67,16 @@
     "await connection.close()\n",
     "# Or: await connection.close(close_connection_pool=False)\n",
     "await connection.connection_pool.disconnect()"
-   ]
-  },
-  {
-   "cell_type": "markdown",
+   ],
    "metadata": {
+    "collapsed": false,
     "pycharm": {
-     "name": "#%% md\n"
+     "name": "#%%\n"
     }
-   },
+   }
+  },
+  {
+   "cell_type": "markdown",
    "source": [
     "## Transactions (Multi/Exec)\n",
     "\n",
@@ -87,16 +85,17 @@
     "The commands will not be reflected in Redis until execute() is called & awaited.\n",
     "\n",
     "Usually, when performing a bulk operation, taking advantage of a “transaction” (e.g., Multi/Exec) is to be desired, as it will also add a layer of atomicity to your bulk operation."
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%% md\n"
+    }
+   }
   },
   {
    "cell_type": "code",
    "execution_count": 3,
-   "metadata": {
-    "pycharm": {
-     "name": "#%%\n"
-    }
-   },
    "outputs": [],
    "source": [
     "import redis.asyncio as redis\n",
@@ -106,25 +105,31 @@
     "    ok1, ok2 = await (pipe.set(\"key1\", \"value1\").set(\"key2\", \"value2\").execute())\n",
     "assert ok1\n",
     "assert ok2"
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
   },
   {
    "cell_type": "markdown",
-   "metadata": {},
    "source": [
     "## Pub/Sub Mode\n",
     "\n",
     "Subscribing to specific channels:"
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%% md\n"
+    }
+   }
   },
   {
    "cell_type": "code",
    "execution_count": 4,
-   "metadata": {
-    "pycharm": {
-     "name": "#%%\n"
-    }
-   },
    "outputs": [
     {
      "name": "stdout",
@@ -165,23 +170,29 @@
     "    await r.publish(\"channel:1\", STOPWORD)\n",
     "\n",
     "    await future"
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
   },
   {
    "cell_type": "markdown",
-   "metadata": {},
    "source": [
     "Subscribing to channels matching a glob-style pattern:"
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%% md\n"
+    }
+   }
   },
   {
    "cell_type": "code",
    "execution_count": 5,
-   "metadata": {
-    "pycharm": {
-     "name": "#%%\n"
-    }
-   },
    "outputs": [
     {
      "name": "stdout",
@@ -223,11 +234,16 @@
     "    await r.publish(\"channel:1\", STOPWORD)\n",
     "\n",
     "    await future"
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
   },
   {
    "cell_type": "markdown",
-   "metadata": {},
    "source": [
     "## Sentinel Client\n",
     "\n",
@@ -236,16 +252,17 @@
     "Calling aioredis.sentinel.Sentinel.master_for or aioredis.sentinel.Sentinel.slave_for methods will return Redis clients connected to specified services monitored by Sentinel.\n",
     "\n",
     "Sentinel client will detect failover and reconnect Redis clients automatically."
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%% md\n"
+    }
+   }
   },
   {
    "cell_type": "code",
    "execution_count": null,
-   "metadata": {
-    "pycharm": {
-     "name": "#%%\n"
-    }
-   },
    "outputs": [],
    "source": [
     "import asyncio\n",
@@ -260,7 +277,13 @@
     "assert ok\n",
     "val = await r.get(\"key\")\n",
     "assert val == b\"value\""
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
   }
  ],
  "metadata": {
@@ -284,4 +307,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 1
-}
+}
\ No newline at end of file
diff --git a/docs/examples/connection_examples.ipynb b/docs/examples/connection_examples.ipynb
index b0084ff..ca8dd44 100644
--- a/docs/examples/connection_examples.ipynb
+++ b/docs/examples/connection_examples.ipynb
@@ -99,6 +99,197 @@
   },
   {
    "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": {},
    "source": [
     "## Connecting to Redis instances by specifying a URL scheme.\n",
@@ -176,4 +367,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 2
-}
+}
\ No newline at end of file