Redis flushdb

ansible_base/lib/redis/client.py

@@ -1,15 +1,15 @@
 import copy
 import logging
 import os
-from typing import Dict, Union
+from typing import Any, Awaitable, Dict, Optional, Union
 from urllib.parse import parse_qs, urlparse
 
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.translation import gettext as _
 from django_redis.client import DefaultClient
 from redis import Redis
 from redis.cluster import ClusterNode, RedisCluster
-from redis.exceptions import RedisClusterException
+from redis.exceptions import NoPermissionError, RedisClusterException
 
 from ansible_base.lib.constants import STATUS_DEGRADED, STATUS_FAILED, STATUS_GOOD
 
@@ -20,11 +20,48 @@
 
 _INVALID_STANDALONE_OPTIONS = ['cluster_error_retry_attempts']
 
+# This will eventually be in the Redis package under redis.typing later
+ResponseT = Union[Awaitable, Any]
+
+
+class DABRedisRespectACLFlushMixin:
+    """
+    FLUSHDB is called when we call django.cache.clear()
+
+    FLUSHDB does NOT respect Redis ACL on keys. It flushes all keys in a database.
+    You think your keys are protected because the user you are running FLUSHDB as
+    has write-only access to keys with the pattern gateway* or even read-only access
+    ... WRONG, you aren't safe!
+
+    And if you are thinking of using Redis' command re-write feature to sweep this
+    problem under the rug then god help the developer or support person that comes after you.
+
+    At some point in the future we may also want to override flushall as well
+    """
+
+    def flushdb(self, asynchronous: Optional[bool] = None, **kwargs) -> ResponseT:
+        if asynchronous is not None:
+            logger.warning("DABRedis clients implement an ACL friendly FLUSHDB which can not be async")
+
+        if self.__class__ == DABRedisCluster:
+            all_keys = self.keys("*", target_nodes=self.ALL_NODES)
+        elif self.__class__ == DABRedis:
+            all_keys = self.keys("*")
+        else:
+            raise ImproperlyConfigured(f"Got an inappropriate type of client {self.__class__}")
+
+        if all_keys:
+            # Only attempt to delete keys if we got some otherwise we will get an exception from the delete function
+            try:
+                self.delete(*all_keys)
+            except NoPermissionError:
+                pass
+
 
 # We are going to build our own cluster class to override the mget function
 # In a redis cluster, keys might not be in the same slot and this will throw off mget.
 # Instead, we are going to try and use mget and then, if we get the slot error, we will try the mget_nonatomic to make it work
-class DABRedisCluster(RedisCluster):
+class DABRedisCluster(DABRedisRespectACLFlushMixin, RedisCluster):
     mode = 'cluster'
 
     def mget(self, *args, **kwargs):
@@ -36,7 +73,7 @@ def mget(self, *args, **kwargs):
             raise
 
 
-class DABRedis(Redis):
+class DABRedis(DABRedisRespectACLFlushMixin, Redis):
     mode = 'standalone'
 
 
@@ -113,29 +150,38 @@ def _get_hosts(self) -> None:
         self.connection_settings.pop('port', None)
         self.connection_settings['startup_nodes'] = []
 
-        translated_generic_exception = ImproperlyConfigured(_('Unable to parse redis_hosts, see logs for more details'))
-
         # Make sure we have a string for redis_hosts
-        if not isinstance(self.redis_hosts, str):
-            logger.error(f"Specified redis_hosts is not a string, got: {self.redis_hosts}")
-            raise translated_generic_exception
-
-        host_ports = self.redis_hosts.split(',')
-        for host_port in host_ports:
-            try:
-                node, port_string = host_port.split(':')
-            except ValueError:
-                logger.error(f"Specified cluster_host {host_port} is not valid; it needs to be in the format <host>:<port>")
-                raise translated_generic_exception
-
-            # Make sure we have an int for the port
-            try:
-                port = int(port_string)
-            except ValueError:
-                logger.error(f'Specified port on {host_port} is not an int')
-                raise translated_generic_exception
-
-            self.connection_settings['startup_nodes'].append(ClusterNode(node, port))
+        try:
+            if not isinstance(self.redis_hosts, str):
+                logger.error(f"Specified redis_hosts is not a string, got: {self.redis_hosts}")
+                # Since we didn't get a string we can't test any further, raise an exception here
+                raise ValueError()
+
+            had_host_errors = False
+            host_ports = self.redis_hosts.split(',')
+            for host_port in host_ports:
+                try:
+                    node, port_string = host_port.split(':')
+                except ValueError:
+                    logger.error(f"Specified cluster_host {host_port} is not valid; it needs to be in the format <host>:<port>")
+                    had_host_errors = True
+                    continue
+
+                # Make sure we have an int for the port
+                try:
+                    port = int(port_string)
+                except ValueError:
+                    logger.error(f'Specified port on {host_port} is not an int')
+                    had_host_errors = True
+                    continue
+
+                self.connection_settings['startup_nodes'].append(ClusterNode(node, port))
+
+            if had_host_errors:
+                raise ValueError()
+
+        except ValueError:
+            raise ImproperlyConfigured(_('Unable to parse redis_hosts, see logs for more details'))
 
     def __init__(self, *args, **kwargs):
         self.url = ''

test_app/tests/lib/cache/test_fallback_cache.py

@@ -97,7 +97,10 @@ def test_fallback_cache():
     # Check until primary is back
     timeout = time.time() + 30
     while True:
-        if cache.get_active_cache() == PRIMARY_CACHE:
+        # Tell the cache to get a key, this should cause it to check the primary cache
+        cache.get("key")
+        active_cache = cache.get_active_cache()
+        if active_cache == PRIMARY_CACHE:
             break
         if time.time() > timeout:
             assert False

test_app/tests/lib/redis/test_client.py

@@ -12,7 +12,9 @@
 from ansible_base.lib.redis.client import (
     _DEFAULT_STATUS_TIMEOUT_SEC,
     _REDIS_CLUSTER_OK_STATUS,
+    DABRedis,
     DABRedisCluster,
+    DABRedisRespectACLFlushMixin,
     RedisClient,
     determine_cluster_node_status,
     get_redis_client,
@@ -396,3 +398,73 @@ def test_redis_standalone_removes_cluster_settings():
         rm.assert_called_once
         assert 'host' in rm.call_args.kwargs
         assert 'cluster_error_retry_attempts' not in rm.call_args.kwargs
+
+
+@pytest.mark.parametrize(
+    "the_class",
+    [
+        DABRedis,
+        DABRedisCluster,
+    ],
+)
+def test_redis_clients_have_flushdb_mixin(the_class):
+    assert issubclass(the_class, DABRedisRespectACLFlushMixin)
+
+
+def test_flushdb_warns_about_async(expected_log):
+    args = {'mode': 'standalone'}
+    from ansible_base.lib.redis.client import RedisClientGetter
+
+    with mock.patch('redis.Redis.__init__', return_value=None):
+        client_getter = RedisClientGetter()
+        client = client_getter.get_client('rediss://localhost', **args)
+        client.keys = mock.MagicMock(return_value=[])
+        client.delete = mock.MagicMock(return_value=None)
+        with expected_log('ansible_base.lib.redis.client.logger', 'warning', "DABRedis clients implement an ACL friendly FLUSHDB which can not be async"):
+            client.flushdb(asynchronous=True)
+            # The delete should not be called in this case because keys returned nothing.
+            client.delete.assert_not_called()
+
+
+class notADABClient(DABRedisRespectACLFlushMixin, Redis):
+    pass
+
+
+def test_flushdb_not_a_dab_client():
+    with mock.patch('redis.Redis.__init__', return_value=None):
+        with pytest.raises(ImproperlyConfigured):
+            client = notADABClient('rediss://localhost')
+            client.flushdb()
+
+
+def test_ensure_flushdb_does_not_die_on_no_permission_exception():
+    args = {'mode': 'standalone'}
+    from redis.exceptions import NoPermissionError
+
+    from ansible_base.lib.redis.client import RedisClientGetter
+
+    with mock.patch('redis.Redis.__init__', return_value=None):
+        client_getter = RedisClientGetter()
+        client = client_getter.get_client('rediss://localhost', **args)
+        client.keys = mock.MagicMock(return_value=["key1"])
+        client.delete = mock.MagicMock(side_effect=NoPermissionError())
+        # This should just not raise if NoPermissionError is raised
+        client.flushdb()
+        # We had a key returned so the delete should have been called with it.
+        client.delete.assert_called()
+
+
+def test_redis_cluster_passes_target_nodes():
+    args = {'mode': 'cluster'}
+    from ansible_base.lib.redis.client import RedisClientGetter
+
+    with mock.patch('redis.RedisCluster.__init__', return_value=None):
+        client_getter = RedisClientGetter()
+        client = client_getter.get_client('rediss://localhost', **args)
+        client.keys = mock.MagicMock(return_value=[])
+        client.delete = mock.MagicMock(return_value=None)
+        # This should just not raise if NoPermissionError is raised
+        client.flushdb()
+        client.keys.assert_called_with("*", target_nodes=client.ALL_NODES)
+        # There were no keys specified so delete should not have been called.
+        client.delete.assert_not_called()