Repository: redis/redis-py
Pull request #2261 (2022-07-05 11:45:41)

CredentialsProvider class added to support password rotation
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Does `$ tox` pass with this change (including linting)?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] Is the new or changed code fully tested?
- [x] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [x] Is there an example added to the examples folder (if applicable)?
- [x] Was the change added to CHANGES file?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

Instead of providing a simple pair of username+password, the users can create a CredentialsProvider object with their own credentials supplier function, and have redis-py call it whenever a new connection is created.
By doing so, users will be able to fetch the current password and rotate credentials without having to create a new client.

See related feature request in Lettuce: https://github.com/lettuce-io/lettuce-core/issues/1774 

codecov-commenter: # [Codecov](https://codecov.io/gh/redis/redis-py/pull/2261?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
> Merging [#2261](https://codecov.io/gh/redis/redis-py/pull/2261?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (63da14e) into [master](https://codecov.io/gh/redis/redis-py/commit/e6cd4fdf3b159d7f118f154e28e884069da89d7e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (e6cd4fd) will **decrease** coverage by `14.66%`.
> The diff coverage is `90.00%`.

```diff
@@             Coverage Diff             @@
##           master    #2261       +/-   ##
===========================================
- Coverage   91.85%   77.19%   -14.67%     
===========================================
  Files         108      108               
  Lines       27783    26976      -807     
===========================================
- Hits        25521    20823     -4698     
- Misses       2262     6153     +3891     
```


| [Impacted Files](https://codecov.io/gh/redis/redis-py/pull/2261?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [redis/\_\_init\_\_.py](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvX19pbml0X18ucHk=) | `76.19% <ø> (-4.77%)` | :arrow_down: |
| [redis/client.py](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY2xpZW50LnB5) | `87.90% <ø> (-0.79%)` | :arrow_down: |
| [redis/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY2x1c3Rlci5weQ==) | `18.71% <ø> (-71.71%)` | :arrow_down: |
| [redis/connection.py](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY29ubmVjdGlvbi5weQ==) | `77.32% <80.64%> (-9.11%)` | :arrow_down: |
| [tests/test\_connection.py](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9jb25uZWN0aW9uLnB5) | `97.74% <95.91%> (-1.14%)` | :arrow_down: |
| [tests/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9jbHVzdGVyLnB5) | `14.66% <0.00%> (-82.31%)` | :arrow_down: |
| [tests/test\_asyncio/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY2x1c3Rlci5weQ==) | `18.62% <0.00%> (-78.82%)` | :arrow_down: |
| [redis/asyncio/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jbHVzdGVyLnB5) | `16.83% <0.00%> (-73.13%)` | :arrow_down: |
| [redis/asyncio/parser.py](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9wYXJzZXIucHk=) | `17.30% <0.00%> (-63.47%)` | :arrow_down: |
| ... and [65 more](https://codecov.io/gh/redis/redis-py/pull/2261/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/redis/redis-py/pull/2261?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/redis/redis-py/pull/2261?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Last update [e6cd4fd...63da14e](https://codecov.io/gh/redis/redis-py/pull/2261?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).

barshaul: Added type hints + moved CredentialsProvider to a separate file
barshaul: @chayim @dvora-h - Finished round, ready for review :). Tests failed here from some internal error, in my fork it seems that tests are passing:
https://github.com/barshaul/redis-py/pull/16#partial-pull-merging
chayim: @barshaul Thanks as always. I left a tonne of comments - and will give it another once over. As part of this, I suggest a MockCredentialProvider or similar. It would be nice to provide an example of a standard provider, perhaps implemented by a single (useless) function. This is how @dvora-h initially looked at this PR in fact.
barshaul: @chayim Finished round. I left some of the reviews open with questions
dvora-h: @barshaul See tests failure 
barshaul: > @barshaul See tests failure

@dvora-h fixed, not sure why the tests are still queued 
chayim: @dvora-h once @barshaul addresses @akx suggestions, I think an end-to-end (re)review of this is worthwhile. It can't make rc3 but could make rc4, potentially.
barshaul: @chayim @dvora-h @akx  - finished round
dvora-h: @barshaul I think that after adding async support we are good to go.
dvora-h: LGTM now!

Patch:
diff --git a/CHANGES b/CHANGES
index 7bdfacf9..4945f616 100644
--- a/CHANGES
+++ b/CHANGES
@@ -24,6 +24,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
     * Enable Lock for asyncio cluster mode
 
 * 4.1.3 (Feb 8, 2022)
diff --git a/docs/examples/asyncio_examples.ipynb b/docs/examples/asyncio_examples.ipynb
index dab7a96a..855255c8 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 b0084ff0..ca8dd443 100644
--- a/docs/examples/connection_examples.ipynb
+++ b/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
-}
+}
\ No newline at end of file
diff --git a/redis/__init__.py b/redis/__init__.py
index b7560a67..5201fe22 100644
--- a/redis/__init__.py
+++ b/redis/__init__.py
@@ -9,6 +9,7 @@ from redis.connection import (
     SSLConnection,
     UnixDomainSocketConnection,
 )
+from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
 from redis.exceptions import (
     AuthenticationError,
     AuthenticationWrongNumberOfArgsError,
@@ -62,6 +63,7 @@ __all__ = [
     "Connection",
     "ConnectionError",
     "ConnectionPool",
+    "CredentialProvider",
     "DataError",
     "from_url",
     "InvalidResponse",
@@ -76,6 +78,7 @@ __all__ = [
     "SentinelManagedConnection",
     "SentinelManagedSSLConnection",
     "SSLConnection",
+    "UsernamePasswordCredentialProvider",
     "StrictRedis",
     "TimeoutError",
     "UnixDomainSocketConnection",
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 619ee111..c0855717 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -46,6 +46,7 @@ from redis.commands import (
     list_or_args,
 )
 from redis.compat import Protocol, TypedDict
+from redis.credentials import CredentialProvider
 from redis.exceptions import (
     ConnectionError,
     ExecAbortError,
@@ -174,6 +175,7 @@ class Redis(
         retry: Optional[Retry] = None,
         auto_close_connection_pool: bool = True,
         redis_connect_func=None,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
         """
         Initialize a new Redis client.
@@ -199,6 +201,7 @@ class Redis(
                 "db": db,
                 "username": username,
                 "password": password,
+                "credential_provider": credential_provider,
                 "socket_timeout": socket_timeout,
                 "encoding": encoding,
                 "encoding_errors": encoding_errors,
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index 97f41518..57aafbd6 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -40,6 +40,7 @@ from redis.cluster import (
 )
 from redis.commands import READ_COMMANDS, AsyncRedisClusterCommands
 from redis.crc import REDIS_CLUSTER_HASH_SLOTS, key_slot
+from redis.credentials import CredentialProvider
 from redis.exceptions import (
     AskError,
     BusyLoadingError,
@@ -220,6 +221,7 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
         # Client related kwargs
         db: Union[str, int] = 0,
         path: Optional[str] = None,
+        credential_provider: Optional[CredentialProvider] = None,
         username: Optional[str] = None,
         password: Optional[str] = None,
         client_name: Optional[str] = None,
@@ -266,6 +268,7 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
             "connection_class": Connection,
             "parser_class": ClusterParser,
             # Client related kwargs
+            "credential_provider": credential_provider,
             "username": username,
             "password": password,
             "client_name": client_name,
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 1288bb67..df066c47 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -29,6 +29,7 @@ import async_timeout
 from redis.asyncio.retry import Retry
 from redis.backoff import NoBackoff
 from redis.compat import Protocol, TypedDict
+from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
 from redis.exceptions import (
     AuthenticationError,
     AuthenticationWrongNumberOfArgsError,
@@ -416,6 +417,7 @@ class Connection:
         "db",
         "username",
         "client_name",
+        "credential_provider",
         "password",
         "socket_timeout",
         "socket_connect_timeout",
@@ -465,14 +467,23 @@ class Connection:
         retry: Optional[Retry] = None,
         redis_connect_func: Optional[ConnectCallbackT] = None,
         encoder_class: Type[Encoder] = Encoder,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
+        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.pid = os.getpid()
         self.host = host
         self.port = int(port)
         self.db = db
-        self.username = username
         self.client_name = client_name
+        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 or None
         self.socket_keepalive = socket_keepalive
@@ -637,14 +648,13 @@ class Connection:
         """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:
-            auth_args: Union[Tuple[str], Tuple[str, str]]
-            if self.username:
-                auth_args = (self.username, self.password or "")
-            else:
-                # Mypy bug: https://github.com/python/mypy/issues/10944
-                auth_args = (self.password or "",)
+        # 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
             await self.send_command("AUTH", *auth_args, check_health=False)
@@ -656,7 +666,7 @@ class Connection:
                 # 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
-                await self.send_command("AUTH", self.password, check_health=False)
+                await self.send_command("AUTH", auth_args[-1], check_health=False)
                 auth_response = await self.read_response()
 
             if str_if_bytes(auth_response) != "OK":
@@ -1014,18 +1024,27 @@ class UnixDomainSocketConnection(Connection):  # lgtm [py/missing-call-to-init]
         client_name: str = None,
         retry: Optional[Retry] = None,
         redis_connect_func=None,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
         """
         Initialize a new UnixDomainSocketConnection.
         To specify a retry policy, first set `retry_on_timeout` to `True`
         then set `retry` to a valid `Retry` object
         """
+        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.pid = os.getpid()
         self.path = path
         self.db = db
-        self.username = username
         self.client_name = client_name
+        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 or None
         self.retry_on_timeout = retry_on_timeout
diff --git a/redis/client.py b/redis/client.py
index 6a26d289..8356ba7d 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -5,6 +5,7 @@ import threading
 import time
 import warnings
 from itertools import chain
+from typing import Optional
 
 from redis.commands import (
     CoreCommands,
@@ -13,6 +14,7 @@ from redis.commands import (
     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 @@ class Redis(AbstractRedis, RedisModuleCommands, CoreCommands, SentinelCommands):
         username=None,
         retry=None,
         redis_connect_func=None,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
         """
         Initialize a new Redis client.
@@ -985,6 +988,7 @@ class Redis(AbstractRedis, RedisModuleCommands, CoreCommands, SentinelCommands):
                 "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:
diff --git a/redis/cluster.py b/redis/cluster.py
index cb3b2a6c..027fe407 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -121,6 +121,7 @@ REDIS_ALLOWED_KEYS = (
     "connection_class",
     "connection_pool",
     "client_name",
+    "credential_provider",
     "db",
     "decode_responses",
     "encoding",
diff --git a/redis/connection.py b/redis/connection.py
index fecb06b0..a2b00740 100755
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -8,9 +8,11 @@ import weakref
 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 @@ class Connection:
         username=None,
         retry=None,
         redis_connect_func=None,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
         """
         Initialize a new Connection.
@@ -510,13 +513,21 @@ class Connection:
         `retry` to a valid `Retry` object.
         To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
+        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.pid = os.getpid()
         self.host = host
         self.port = int(port)
         self.db = db
-        self.username = username
         self.client_name = client_name
+        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 +686,13 @@ class Connection:
         "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 +704,7 @@ class Connection:
                 # 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 +1062,7 @@ class UnixDomainSocketConnection(Connection):
         client_name=None,
         retry=None,
         redis_connect_func=None,
+        credential_provider: Optional[CredentialProvider] = None,
     ):
         """
         Initialize a new UnixDomainSocketConnection.
@@ -1058,12 +1071,20 @@ class UnixDomainSocketConnection(Connection):
         `retry` to a valid `Retry` object.
         To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
+        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.pid = os.getpid()
         self.path = path
         self.db = db
-        self.username = username
         self.client_name = client_name
+        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:
diff --git a/redis/credentials.py b/redis/credentials.py
new file mode 100644
index 00000000..7ba26dcd
--- /dev/null
+++ b/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,)


Test patch:
diff --git a/tests/test_asyncio/test_credentials.py b/tests/test_asyncio/test_credentials.py
new file mode 100644
index 00000000..8e213cdb
--- /dev/null
+++ b/tests/test_asyncio/test_credentials.py
@@ -0,0 +1,284 @@
+import functools
+import random
+import string
+from typing import Optional, Tuple, Union
+
+import pytest
+import pytest_asyncio
+
+import redis
+from redis import AuthenticationError, DataError, ResponseError
+from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
+from redis.utils import str_if_bytes
+from tests.conftest import skip_if_redis_enterprise
+
+
+@pytest_asyncio.fixture()
+async def r_acl_teardown(r: redis.Redis):
+    """
+    A special fixture which removes the provided names from the database after use
+    """
+    usernames = []
+
+    def factory(username):
+        usernames.append(username)
+        return r
+
+    yield factory
+    for username in usernames:
+        await r.acl_deluser(username)
+
+
+@pytest_asyncio.fixture()
+async def r_required_pass_teardown(r: redis.Redis):
+    """
+    A special fixture which removes the provided password from the database after use
+    """
+    passwords = []
+
+    def factory(username):
+        passwords.append(username)
+        return r
+
+    yield factory
+    for password in passwords:
+        try:
+            await r.auth(password)
+        except (ResponseError, AuthenticationError):
+            await r.auth("default", "")
+        await r.config_set("requirepass", "")
+
+
+class NoPassCredProvider(CredentialProvider):
+    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:
+        return "username", ""
+
+
+class AsyncRandomAuthCredProvider(CredentialProvider):
+    def __init__(self, user: Optional[str], endpoint: str):
+        self.user = user
+        self.endpoint = endpoint
+
+    @functools.lru_cache(maxsize=10)
+    def get_credentials(self) -> Union[Tuple[str, str], Tuple[str]]:
+        def get_random_string(length):
+            letters = string.ascii_lowercase
+            result_str = "".join(random.choice(letters) for i in range(length))
+            return result_str
+
+        if self.user:
+            auth_token: str = get_random_string(5) + self.user + "_" + self.endpoint
+            return self.user, auth_token
+        else:
+            auth_token: str = get_random_string(5) + self.endpoint
+            return (auth_token,)
+
+
+async def init_acl_user(r, username, password):
+    # reset the user
+    await r.acl_deluser(username)
+    if password:
+        assert (
+            await r.acl_setuser(
+                username,
+                enabled=True,
+                passwords=["+" + password],
+                keys="~*",
+                commands=[
+                    "+ping",
+                    "+command",
+                    "+info",
+                    "+select",
+                    "+flushdb",
+                    "+cluster",
+                ],
+            )
+            is True
+        )
+    else:
+        assert (
+            await r.acl_setuser(
+                username,
+                enabled=True,
+                keys="~*",
+                commands=[
+                    "+ping",
+                    "+command",
+                    "+info",
+                    "+select",
+                    "+flushdb",
+                    "+cluster",
+                ],
+                nopass=True,
+            )
+            is True
+        )
+
+
+async def init_required_pass(r, password):
+    await r.config_set("requirepass", password)
+
+
+@pytest.mark.asyncio
+class TestCredentialsProvider:
+    @skip_if_redis_enterprise()
+    async def test_only_pass_without_creds_provider(
+        self, r_required_pass_teardown, create_redis
+    ):
+        # test for default user (`username` is supposed to be optional)
+        password = "password"
+        r = r_required_pass_teardown(password)
+        await init_required_pass(r, password)
+        assert await r.auth(password) is True
+
+        r2 = await create_redis(flushdb=False, password=password)
+
+        assert await r2.ping() is True
+
+    @skip_if_redis_enterprise()
+    async def test_user_and_pass_without_creds_provider(
+        self, r_acl_teardown, create_redis
+    ):
+        """
+        Test backward compatibility with username and password
+        """
+        # test for other users
+        username = "username"
+        password = "password"
+        r = r_acl_teardown(username)
+        await init_acl_user(r, username, password)
+        r2 = await create_redis(flushdb=False, username=username, password=password)
+
+        assert await r2.ping() is True
+
+    @pytest.mark.parametrize("username", ["username", None])
+    @skip_if_redis_enterprise()
+    @pytest.mark.onlynoncluster
+    async def test_credential_provider_with_supplier(
+        self, r_acl_teardown, r_required_pass_teardown, create_redis, username
+    ):
+        creds_provider = AsyncRandomAuthCredProvider(
+            user=username,
+            endpoint="localhost",
+        )
+
+        auth_args = creds_provider.get_credentials()
+        password = auth_args[-1]
+
+        if username:
+            r = r_acl_teardown(username)
+            await init_acl_user(r, username, password)
+        else:
+            r = r_required_pass_teardown(password)
+            await init_required_pass(r, password)
+
+        r2 = await create_redis(flushdb=False, credential_provider=creds_provider)
+
+        assert await r2.ping() is True
+
+    async def test_async_credential_provider_no_password_success(
+        self, r_acl_teardown, create_redis
+    ):
+        username = "username"
+        r = r_acl_teardown(username)
+        await init_acl_user(r, username, "")
+        r2 = await create_redis(
+            flushdb=False,
+            credential_provider=NoPassCredProvider(),
+        )
+        assert await r2.ping() is True
+
+    @pytest.mark.onlynoncluster
+    async def test_credential_provider_no_password_error(
+        self, r_acl_teardown, create_redis
+    ):
+        username = "username"
+        r = r_acl_teardown(username)
+        await init_acl_user(r, username, "password")
+        with pytest.raises(AuthenticationError) as e:
+            await create_redis(
+                flushdb=False,
+                credential_provider=NoPassCredProvider(),
+                single_connection_client=True,
+            )
+        assert e.match("invalid username-password")
+        assert await r.acl_deluser(username)
+
+    @pytest.mark.onlynoncluster
+    async def test_password_and_username_together_with_cred_provider_raise_error(
+        self, r_acl_teardown, create_redis
+    ):
+        username = "username"
+        r = r_acl_teardown(username)
+        await init_acl_user(r, username, "password")
+        cred_provider = UsernamePasswordCredentialProvider(
+            username="username", password="password"
+        )
+        with pytest.raises(DataError) as e:
+            await create_redis(
+                flushdb=False,
+                username="username",
+                password="password",
+                credential_provider=cred_provider,
+                single_connection_client=True,
+            )
+        assert e.match(
+            "'username' and 'password' cannot be passed along with "
+            "'credential_provider'."
+        )
+
+    @pytest.mark.onlynoncluster
+    async def test_change_username_password_on_existing_connection(
+        self, r_acl_teardown, create_redis
+    ):
+        username = "origin_username"
+        password = "origin_password"
+        new_username = "new_username"
+        new_password = "new_password"
+        r = r_acl_teardown(username)
+        await init_acl_user(r, username, password)
+        r2 = await create_redis(flushdb=False, username=username, password=password)
+        assert await r2.ping() is True
+        conn = await r2.connection_pool.get_connection("_")
+        await conn.send_command("PING")
+        assert str_if_bytes(await conn.read_response()) == "PONG"
+        assert conn.username == username
+        assert conn.password == password
+        await init_acl_user(r, new_username, new_password)
+        conn.password = new_password
+        conn.username = new_username
+        await conn.send_command("PING")
+        assert str_if_bytes(await conn.read_response()) == "PONG"
+
+
+@pytest.mark.asyncio
+class TestUsernamePasswordCredentialProvider:
+    async def test_user_pass_credential_provider_acl_user_and_pass(
+        self, r_acl_teardown, create_redis
+    ):
+        username = "username"
+        password = "password"
+        r = r_acl_teardown(username)
+        provider = UsernamePasswordCredentialProvider(username, password)
+        assert provider.username == username
+        assert provider.password == password
+        assert provider.get_credentials() == (username, password)
+        await init_acl_user(r, provider.username, provider.password)
+        r2 = await create_redis(flushdb=False, credential_provider=provider)
+        assert await r2.ping() is True
+
+    async def test_user_pass_provider_only_password(
+        self, r_required_pass_teardown, create_redis
+    ):
+        password = "password"
+        provider = UsernamePasswordCredentialProvider(password=password)
+        r = r_required_pass_teardown(password)
+        assert provider.username == ""
+        assert provider.password == password
+        assert provider.get_credentials() == (password,)
+
+        await init_required_pass(r, password)
+
+        r2 = await create_redis(flushdb=False, credential_provider=provider)
+        assert await r2.auth(provider.password) is True
+        assert await r2.ping() is True
diff --git a/tests/test_credentials.py b/tests/test_credentials.py
new file mode 100644
index 00000000..9aeb1ef1
--- /dev/null
+++ b/tests/test_credentials.py
@@ -0,0 +1,245 @@
+import functools
+import random
+import string
+from typing import Optional, Tuple, Union
+
+import pytest
+
+import redis
+from redis import AuthenticationError, DataError, ResponseError
+from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
+from redis.utils import str_if_bytes
+from tests.conftest import _get_client, skip_if_redis_enterprise
+
+
+class NoPassCredProvider(CredentialProvider):
+    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:
+        return "username", ""
+
+
+class RandomAuthCredProvider(CredentialProvider):
+    def __init__(self, user: Optional[str], endpoint: str):
+        self.user = user
+        self.endpoint = endpoint
+
+    @functools.lru_cache(maxsize=10)
+    def get_credentials(self) -> Union[Tuple[str, str], Tuple[str]]:
+        def get_random_string(length):
+            letters = string.ascii_lowercase
+            result_str = "".join(random.choice(letters) for i in range(length))
+            return result_str
+
+        if self.user:
+            auth_token: str = get_random_string(5) + self.user + "_" + self.endpoint
+            return self.user, auth_token
+        else:
+            auth_token: str = get_random_string(5) + self.endpoint
+            return (auth_token,)
+
+
+def init_acl_user(r, request, username, password):
+    # reset the user
+    r.acl_deluser(username)
+    if password:
+        assert (
+            r.acl_setuser(
+                username,
+                enabled=True,
+                passwords=["+" + password],
+                keys="~*",
+                commands=[
+                    "+ping",
+                    "+command",
+                    "+info",
+                    "+select",
+                    "+flushdb",
+                    "+cluster",
+                ],
+            )
+            is True
+        )
+    else:
+        assert (
+            r.acl_setuser(
+                username,
+                enabled=True,
+                keys="~*",
+                commands=[
+                    "+ping",
+                    "+command",
+                    "+info",
+                    "+select",
+                    "+flushdb",
+                    "+cluster",
+                ],
+                nopass=True,
+            )
+            is True
+        )
+
+    if request is not None:
+
+        def teardown():
+            r.acl_deluser(username)
+
+        request.addfinalizer(teardown)
+
+
+def init_required_pass(r, request, password):
+    r.config_set("requirepass", password)
+
+    def teardown():
+        try:
+            r.auth(password)
+        except (ResponseError, AuthenticationError):
+            r.auth("default", "")
+        r.config_set("requirepass", "")
+
+    request.addfinalizer(teardown)
+
+
+class TestCredentialsProvider:
+    @skip_if_redis_enterprise()
+    def test_only_pass_without_creds_provider(self, r, request):
+        # test for default user (`username` is supposed to be optional)
+        password = "password"
+        init_required_pass(r, request, password)
+        assert r.auth(password) is True
+
+        r2 = _get_client(redis.Redis, request, flushdb=False, password=password)
+
+        assert r2.ping() is True
+
+    @skip_if_redis_enterprise()
+    def test_user_and_pass_without_creds_provider(self, r, request):
+        """
+        Test backward compatibility with username and password
+        """
+        # test for other users
+        username = "username"
+        password = "password"
+
+        init_acl_user(r, request, username, password)
+        r2 = _get_client(
+            redis.Redis, request, flushdb=False, username=username, password=password
+        )
+
+        assert r2.ping() is True
+
+    @pytest.mark.parametrize("username", ["username", None])
+    @skip_if_redis_enterprise()
+    @pytest.mark.onlynoncluster
+    def test_credential_provider_with_supplier(self, r, request, username):
+        creds_provider = RandomAuthCredProvider(
+            user=username,
+            endpoint="localhost",
+        )
+
+        password = creds_provider.get_credentials()[-1]
+
+        if username:
+            init_acl_user(r, request, username, password)
+        else:
+            init_required_pass(r, request, password)
+
+        r2 = _get_client(
+            redis.Redis, request, flushdb=False, credential_provider=creds_provider
+        )
+
+        assert r2.ping() is True
+
+    def test_credential_provider_no_password_success(self, r, request):
+        init_acl_user(r, request, "username", "")
+        r2 = _get_client(
+            redis.Redis,
+            request,
+            flushdb=False,
+            credential_provider=NoPassCredProvider(),
+        )
+        assert r2.ping() is True
+
+    @pytest.mark.onlynoncluster
+    def test_credential_provider_no_password_error(self, r, request):
+        init_acl_user(r, request, "username", "password")
+        with pytest.raises(AuthenticationError) as e:
+            _get_client(
+                redis.Redis,
+                request,
+                flushdb=False,
+                credential_provider=NoPassCredProvider(),
+            )
+        assert e.match("invalid username-password")
+
+    @pytest.mark.onlynoncluster
+    def test_password_and_username_together_with_cred_provider_raise_error(
+        self, r, request
+    ):
+        init_acl_user(r, request, "username", "password")
+        cred_provider = UsernamePasswordCredentialProvider(
+            username="username", password="password"
+        )
+        with pytest.raises(DataError) as e:
+            _get_client(
+                redis.Redis,
+                request,
+                flushdb=False,
+                username="username",
+                password="password",
+                credential_provider=cred_provider,
+            )
+        assert e.match(
+            "'username' and 'password' cannot be passed along with "
+            "'credential_provider'."
+        )
+
+    @pytest.mark.onlynoncluster
+    def test_change_username_password_on_existing_connection(self, r, request):
+        username = "origin_username"
+        password = "origin_password"
+        new_username = "new_username"
+        new_password = "new_password"
+        init_acl_user(r, request, username, password)
+        r2 = _get_client(
+            redis.Redis, request, flushdb=False, username=username, password=password
+        )
+        assert r2.ping() is True
+        conn = r2.connection_pool.get_connection("_")
+        conn.send_command("PING")
+        assert str_if_bytes(conn.read_response()) == "PONG"
+        assert conn.username == username
+        assert conn.password == password
+        init_acl_user(r, request, new_username, new_password)
+        conn.password = new_password
+        conn.username = new_username
+        conn.send_command("PING")
+        assert str_if_bytes(conn.read_response()) == "PONG"
+
+
+class TestUsernamePasswordCredentialProvider:
+    def test_user_pass_credential_provider_acl_user_and_pass(self, r, request):
+        username = "username"
+        password = "password"
+        provider = UsernamePasswordCredentialProvider(username, password)
+        assert provider.username == username
+        assert provider.password == password
+        assert provider.get_credentials() == (username, password)
+        init_acl_user(r, request, provider.username, provider.password)
+        r2 = _get_client(
+            redis.Redis, request, flushdb=False, credential_provider=provider
+        )
+        assert r2.ping() is True
+
+    def test_user_pass_provider_only_password(self, r, request):
+        password = "password"
+        provider = UsernamePasswordCredentialProvider(password=password)
+        assert provider.username == ""
+        assert provider.password == password
+        assert provider.get_credentials() == (password,)
+
+        init_required_pass(r, request, password)
+
+        r2 = _get_client(
+            redis.Redis, request, flushdb=False, credential_provider=provider
+        )
+        assert r2.auth(provider.password) is True
+        assert r2.ping() is True