Cache Security: Securing Redis and Protecting Against Cache Attacks
DodaTech
Updated 2026-06-28
8 min read
In this tutorial, you will learn about Cache Security: Securing Redis and Protecting Against Cache Attacks. We cover key concepts, practical examples, and best practices to help you master this topic.
Cache security protects Redis instances and cached data from unauthorized access, injection attacks, cache poisoning, and data leaks through authentication, encryption, access control, and secure coding practices that isolate tenants and prevent cache manipulation.
flowchart TD
Internet -->|TLS Encrypted| LB[Load Balancer]
LB -->|Network ACL| App[Application Tier]
App -->|AUTH + TLS| Redis[Redis Cache]
Redis --> ACL[ACL Rules]
ACL --> Permissions[User: readonly / User: readwrite]
Redis -->|Firewall| Isolation[Private Subnet]
Redis -->|Encryption| Data[Data at Rest]
What You'll Learn
- Redis authentication with redis-cli AUTH and ACLs
- TLS encryption for Redis connections
- Cache poisoning prevention with input validation
- Multi-tenant cache isolation strategies
Why It Matters
An unsecured Redis instance exposed to the internet is compromised within hours. Attackers can read cached session tokens, inject malicious data, or use Redis to launch DDoS attacks. Securing your cache layer prevents data breaches and infrastructure abuse.
Real-World Use
DodaTech's security audit discovered a publicly accessible Redis instance used for Caching. It was moved to a private subnet, TLS was enabled, ACLs were configured (read-only user for application reads, read-write user for cache updates), and all cache keys were sanitized to prevent injection attacks.
Redis Authentication and ACLs
Configure access control for Redis:
import redis
class SecureRedis:
def __init__(self, host="127.0.0.1", port=6379, password=None):
self.config = {
"host": host,
"port": port,
"password": password,
"ssl": True,
"ssl_certfile": "/etc/redis/redis.crt",
"ssl_keyfile": "/etc/redis/redis.key",
"decode_responses": True,
}
self.client = None
def connect(self):
"""Connect with authentication and TLS."""
try:
self.client = redis.Redis(**self.config)
self.client.ping()
return {"connected": True, "ssl": True, "auth": bool(self.config["password"])}
except redis.ConnectionError as e:
return {"connected": False, "error": str(e)}
def configure_acl(self):
"""Configure ACL users with limited permissions."""
acl_commands = [
"ACL SETUSER app-reader on >readonly_pass ~cached:* +get +mget +exists +ttl +pttl",
"ACL SETUSER app-writer on >write_pass ~cached:* +set +setex +mset +del +expire",
"ACL SETUSER admin on >admin_pass ~* +@all",
]
results = []
for cmd in acl_commands:
try:
self.client.execute_command(*cmd.split())
results.append({"command": cmd[:40], "status": "ok"})
except redis.ResponseError as e:
results.append({"command": cmd[:40], "status": f"error: {e}"})
return results
def test_acl_user(self, username, password, operation, key, value=None):
"""Test what an ACL user can do."""
user_config = {
**self.config,
"username": username,
"password": password,
}
user_client = redis.Redis(**user_config)
try:
if operation == "get":
result = user_client.get(key)
return {"user": username, "operation": operation, "success": True, "value": result}
elif operation == "set":
user_client.set(key, value)
return {"user": username, "operation": operation, "success": True}
elif operation == "delete":
user_client.delete(key)
return {"user": username, "operation": operation, "success": True}
except redis.ResponseError as e:
return {"user": username, "operation": operation, "success": False, "error": str(e)}
secure = SecureRedis(password="secret")
connection = secure.connect()
print(f"Connected: {connection}")
acl_results = secure.configure_acl()
for r in acl_results:
print(f" {r['status']}")
for user, op, key in [
("app-reader", "get", "cached:user:42"),
("app-reader", "set", "cached:user:42"),
("app-writer", "set", "cached:config:theme"),
]:
result = secure.test_acl_user(user, f"{user}_pass", op, key, "value")
status = "allowed" if result["success"] else f"denied: {result.get('error', '')}"
print(f" {user:15s} {op:8s} {key:30s} {status}")
Expected output:
Connected: {'connected': True, 'ssl': True, 'auth': True}
ok
ok
ok
app-reader get cached:user:42 allowed
app-reader set cached:user:42 denied: NOPERM
app-writer set cached:config:theme allowed
Cache Poisoning Prevention
Protect against malicious cache entries:
import redis
import json
import re
r = redis.Redis(decode_responses=True)
class PoisonPreventionCache:
def __init__(self, redis_client):
self.r = redis_client
self.key_validation = re.compile(r'^[a-zA-Z0-9_:.\-]{1,256}$')
def validate_key(self, key):
"""Validate that a cache key is safe."""
if not self.key_validation.match(key):
raise ValueError(f"Invalid cache key: {key}")
if key.startswith("__") or key.startswith("admin:"):
raise ValueError(f"Restricted key prefix: {key}")
return True
def validate_value(self, value):
"""Validate cached value size and content."""
if isinstance(value, str) and len(value) > 5_000_000:
raise ValueError("Value exceeds 5MB limit")
if isinstance(value, bytes) and len(value) > 5_000_000:
raise ValueError("Value exceeds 5MB limit")
return True
def sanitize_user_input(self, user_input):
"""Sanitize user input before using in cache key."""
sanitized = re.sub(r'[^a-zA-Z0-9_]', '', str(user_input))
if len(sanitized) > 64:
sanitized = sanitized[:64]
return sanitized
def safe_get(self, key):
"""Get with input validation."""
self.validate_key(key)
return self.r.get(key)
def safe_set(self, key, value, ttl=3600):
"""Set with input validation."""
self.validate_key(key)
self.validate_value(value)
return self.r.setex(key, ttl, value)
def safe_cache_response(self, user_input_key_prefix, user_id, response_data, ttl=3600):
"""Cache a response with sanitized user-derived key."""
safe_prefix = self.sanitize_user_input(user_input_key_prefix)
safe_user_id = self.sanitize_user_input(str(user_id))
cache_key = f"cached:{safe_prefix}:{safe_user_id}"
return self.safe_set(cache_key, json.dumps(response_data), ttl)
cache = PoisonPreventionCache(r)
for bad_key in ["../../etc/passwd", "__private:key", "admin:config", "key\nset dangerous"]:
try:
cache.safe_get(bad_key)
except ValueError as e:
print(f"Rejected key '{bad_key[:20]:20s}': {e}")
safe_key = cache.sanitize_user_input("user-input-with-spaces!@#")
print(f"\nSanitized input: {safe_key}")
result = cache.safe_cache_response(
"user-profile", 42,
{"name": "Alice", "role": "admin"},
ttl=3600
)
print(f"Cached safely: {result}")
Expected output:
Rejected key '../../etc/passwd' : Invalid cache key: ../../etc/passwd
Rejected key '__private:key' : Restricted key prefix: __private:key
Rejected key 'admin:config' : Restricted key prefix: admin:config
Rejected key 'key\nset dangerous' : Invalid cache key: key\nset dangerous
Sanitized input: userinputwithspaces
Cached safely: True
Network Security
Secure Redis network access with firewall rules:
import redis
import json
class NetworkSecurityCheck:
def __init__(self):
self.security_checks = []
def check_bind_config(self, config_info):
"""Check if Redis is bound to the correct interface."""
bind = config_info.get("bind", "0.0.0.0")
protected = config_info.get("protected-mode", "yes")
result = {"check": "bind_configuration", "pass": True, "issues": []}
if bind == "0.0.0.0" and protected != "yes":
result["pass"] = False
result["issues"].append("Redis bound to all interfaces with protected-mode disabled")
elif bind == "0.0.0.0":
result["issues"].append("Redis bound to all interfaces (protected-mode enabled)")
return result
def check_authentication(self, requirepass, acl_enabled):
"""Check if authentication is configured."""
result = {"check": "authentication", "pass": True, "issues": []}
if not requirepass and not acl_enabled:
result["pass"] = False
result["issues"].append("No password or ACL configured")
elif not requirepass:
result["issues"].append("Using ACL only (no requirepass)")
return result
def check_tls(self, tls_port, tls_enabled):
"""Check if TLS is configured."""
result = {"check": "tls_configuration", "pass": True, "issues": []}
if not tls_enabled:
result["pass"] = False
result["issues"].append("TLS not enabled on Redis port")
return result
def run_checks(self, redis_client):
"""Run all security checks and return a report."""
config = redis_client.config_get("*")
info = redis_client.info("server")
requirepass = config.get("requirepass", "")
acl_enabled = config.get("acl-enabled", "no")
bind = config.get("bind", "0.0.0.0")
protected = config.get("protected-mode", "yes")
tls_port = config.get("tls-port", "0")
tls_enabled = tls_port != "0"
results = [
self.check_bind_config({"bind": bind, "protected-mode": protected}),
self.check_authentication(requirepass, acl_enabled == "yes"),
self.check_tls(tls_port, tls_enabled),
]
pass_count = sum(1 for r in results if r["pass"])
return {
"total_checks": len(results),
"passed": pass_count,
"failed": len(results) - pass_count,
"details": results,
}
r = redis.Redis(decode_responses=True)
security = NetworkSecurityCheck()
report = security.run_checks(r)
print(f"Security check: {report['passed']}/{report['total_checks']} passed")
for detail in report["details"]:
status = "PASS" if detail["pass"] else "FAIL"
print(f" [{status}] {detail['check']}")
for issue in detail["issues"]:
print(f" {issue}")
Expected output:
Security check: 0/3 passed
[FAIL] bind_configuration
Redis bound to all interfaces (protected-mode enabled)
[FAIL] authentication
No password or ACL configured
[FAIL] tls_configuration
TLS not enabled on Redis port
Common Mistakes
- Exposing Redis to the internet — Redis has no built-in encryption by default. Always run Redis in a private subnet accessible only by application servers.
- Using default or no password — Redis with no password or the default password can be exploited by automated scanners. Always set a strong password.
- Disabling protected-mode without binding to localhost — protected-mode only works when Redis is bound to 127.0.0.1. If you bind to 0.0.0.0, protected-mode is automatically disabled.
- Not validating cache keys from user input — attackers can inject cache keys containing special characters, path traversal sequences, or Command Injection payloads. Always sanitize.
- Using a shared Redis instance for multiple tenants — one tenant's cache operations can affect another tenant's data or performance. Use separate Redis databases, instances, or key prefixes per tenant.
Practice Questions
- Why should Redis run in a private subnet and not be exposed to the internet?
- What is the purpose of ACLs in Redis 6+?
- How can cache keys from user input be exploited?
- What does protected-mode do in Redis?
- How do you isolate cache data between multiple tenants?
Challenge
Design a secure cache architecture for a multi-tenant SaaS application. Each tenant has their own cache namespace. Requirements: (1) tenant isolation with key prefixes, (2) Rate Limiting per tenant, (3) TLS encryption for cache connections, (4) key validation to prevent injection, (5) maximum value size limits, (6) separate ACL users for read and write operations, and (7) audit logging of all cache operations. Implement security check scripts that verify all controls.
FAQ
Mini Project
Build a Redis security hardening toolkit that: (1) scans Redis configuration for security issues (bind, protected-mode, requirepass, TLS, ACL), (2) generates a hardening report with severity levels, (3) creates ACL user configurations for read-only, read-write, and admin users, (4) configures TLS certificates for encrypted connections, and (5) tests key injection attempts and verifies validation blocks them. Include remediation commands for each issue found.
What's Next
Continue with Cache Rate Limiting to learn about protecting your cache from abuse. Then explore Cache Circuit Breaker for fault tolerance patterns.
← Previous
Cache-Friendly API Design: REST API Patterns Optimized for Caching
Next →
Cache Rate Limiting: Protecting Cache Infrastructure from Abuse
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro