diff --git a/redis/cluster.py b/redis/cluster.py
index b8d6b1997f..4742f6e186 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -227,6 +227,7 @@ class RedisCluster(RedisClusterCommands):
                 "ACL SETUSER",
                 "ACL USERS",
                 "ACL WHOAMI",
+                "AUTH",
                 "CLIENT LIST",
                 "CLIENT SETNAME",
                 "CLIENT GETNAME",
diff --git a/redis/commands/core.py b/redis/commands/core.py
index 968b7cc412..c296d36b79 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -369,14 +369,16 @@ class ManagementCommands(CommandsProtocol):
     Redis management commands
     """
 
-    def auth(self):
+    def auth(self, password, username=None, **kwargs):
         """
-        This function throws a NotImplementedError since it is intentionally
-        not supported.
+        Authenticates the user. If you do not pass username, Redis will try to
+        authenticate for the "default" user. If you do pass username, it will
+        authenticate for the given user.
+        For more information check https://redis.io/commands/auth
         """
-        raise NotImplementedError(
-            "AUTH is intentionally not implemented in the client."
-        )
+        if username:
+            return self.execute_command("AUTH", username, password, **kwargs)
+        return self.execute_command
 
     def bgrewriteaof(self, **kwargs):
         """Tell the Redis server to rewrite the AOF file from data in memory.
diff --git a/tests/test_commands.py b/tests/test_commands.py
index 5c32d5f296..dc10cde165 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -65,16 +65,32 @@ def test_case_insensitive_command_names(self, r):
 
 
 class TestRedisCommands:
+    def test_auth(self, r, request):
+        username = "redis-py-auth"
+
+        def teardown():
+            r.acl_deluser(username)
+
+        request.addfinalizer(teardown)
+
+        assert r.acl_setuser(
+            username,
+            enabled=True,
+            passwords=["+strong_password"],
+            commands=["+acl"],
+        )
+
+        assert r.auth(username=username, password="strong_password") is True
+
+        with pytest.raises(exceptions.ResponseError):
+            r.auth(username=username, password="wrong_password")
+
     def test_command_on_invalid_key_type(self, r):
         r.lpush("a", "1")
         with pytest.raises(redis.ResponseError):
             r["a"]
 
     # SERVER INFORMATION
-    def test_auth_not_implemented(self, r):
-        with pytest.raises(NotImplementedError):
-            r.auth()
-
     @skip_if_server_version_lt("6.0.0")
     def test_acl_cat_no_category(self, r):
         categories = r.acl_cat()