建站
咨詢
售后
建站咨詢
新聞中心
Redis作為一種高性能的內(nèi)存數(shù)據(jù)庫，被廣泛應(yīng)用于緩存和數(shù)據(jù)存儲等場景中。然而，Redis在某些場景下可能會遭遇穿透攻擊，導(dǎo)致緩存失效，造成性能和安全問題。本文將研究Redis穿透技術(shù)的有效性，探討如何應(yīng)對這種攻擊。
一、Redis穿透攻擊
Redis穿透攻擊顧名思義，就是指攻擊者通過惡意構(gòu)造的數(shù)據(jù)請求，繞過緩存層直接訪問數(shù)據(jù)庫，從而耗盡系統(tǒng)資源，導(dǎo)致系統(tǒng)崩潰。常見的攻擊方式包括使用不存在的KEY進(jìn)行查詢、使用特殊字符進(jìn)行查詢等。
Redis穿透攻擊會對系統(tǒng)性能和安全造成嚴(yán)重影響，具體表現(xiàn)在以下幾個方面：
1. 頻繁請求導(dǎo)致緩存穿透：攻擊者通過不斷發(fā)起請求，如果其中存在未被緩存的數(shù)據(jù)，就會導(dǎo)致緩存穿透，從而使得Redis頻繁從數(shù)據(jù)庫中查詢數(shù)據(jù)，造成性能上的損耗和資源浪費(fèi)。
2. 惡意構(gòu)造數(shù)據(jù)請求導(dǎo)致數(shù)據(jù)庫壓力增大：攻擊者通過惡意構(gòu)造的數(shù)據(jù)請求，繞過了Redis緩存層，直接訪問數(shù)據(jù)庫，從而導(dǎo)致數(shù)據(jù)庫的壓力增大，可能會導(dǎo)致數(shù)據(jù)庫崩潰。
3. 安全隱患：攻擊者可以通過不存在的key進(jìn)行查詢，從而得到一些機(jī)密信息，造成嚴(yán)重的安全問題。
二、常見的解決方案及其局限性
針對Redis穿透攻擊，目前主要有以下幾種解決方案：
1. Bloom Filter
Bloom Filter是一種經(jīng)典的高效數(shù)據(jù)結(jié)構(gòu)，可以用來過濾掉一些不可能存在的數(shù)據(jù)，從而避免頻繁的查詢數(shù)據(jù)庫。在Redis中，可以使用RedisBloom module實現(xiàn)Bloom Filter功能。
但是，Bloom Filter也存在著一些局限性。由于布隆過濾器是一個不可靠的近似算法，并不能保證100%的準(zhǔn)確性。如果判定為不存在于布隆過濾器中的數(shù)據(jù)，仍然有可能存在與數(shù)據(jù)庫中，從而無法完全避免Redis穿透攻擊。
2. Cache Aside
Cache Aside是一種常見的緩存更新策略，當(dāng)緩存中不存在數(shù)據(jù)時，從數(shù)據(jù)庫中查詢，然后將查詢結(jié)果存儲到緩存中。如果緩存中存在數(shù)據(jù)，則直接從緩存中返回數(shù)據(jù)。這種方式可以減輕Redis壓力，但仍無法避免Redis穿透攻擊。
3. 限流
限流是一種有效的防范穿透攻擊的策略，通過限制訪問頻率，避免惡意請求對系統(tǒng)造成影響。但是，由于穿透攻擊的特殊性，惡意攻擊者可以通過改變請求的方式或者使用代理IP等方式來規(guī)避限流措施。
三、研究Redis穿透技術(shù)的有效性
針對以上解決方案的局限性，本文將研究Redis穿透技術(shù)的有效性，探討如何應(yīng)對這種攻擊。具體步驟如下：
1. 模擬攻擊
需要模擬穿透攻擊環(huán)境，構(gòu)造一個大量不存在于緩存中的請求?？梢允褂肞ython語言編寫代碼，使用Redis-py庫進(jìn)行Redis連接和操作，同時使用faker庫構(gòu)造一些隨機(jī)字符串作為key。
其中，如下代碼用于構(gòu)造隨機(jī)字符串：
from faker import Faker
fake = Faker()
key = fake.word()
2. 實現(xiàn)布隆過濾器
接下來，需要實現(xiàn)布隆過濾器?？梢允褂肞ython的bitarray庫實現(xiàn)一個簡單的布隆過濾器，具體代碼如下：
from bitarray import bitarray
import mmh3
class BloomFilter:
def __init__(self, capacity, false_positive_prob):
self.capacity = capacity
self.false_positive_prob = false_positive_prob
self.bf_size = self.get_size(capacity, false_positive_prob)
self.num_hashes = self.get_num_hashes(self.bf_size, capacity)
self.bit_arr = bitarray(self.bf_size)
self.bit_arr.setall(0)
def add(self, key):
for i in range(self.num_hashes):
digest = mmh3.hash(key, i) % self.bf_size
self.bit_arr[digest] = 1
def __contns__(self, key):
for i in range(self.num_hashes):
digest = mmh3.hash(key, i) % self.bf_size
if not self.bit_arr[digest]:
return False
return True
def get_size(self, n, p):
m = -(n * math.log(p))/(math.log(2)**2)
return int(m)
def get_num_hashes(self, m, n):
k = (m/n) * math.log(2)
return int(k)
在具體實現(xiàn)中，使用MurmurHash3算法來生成一個key的hash值，并使用這個值對位圖上的對應(yīng)位置進(jìn)行標(biāo)記。當(dāng)查找一個key的時候，同樣使用MurmurHash3算法對key進(jìn)行hash，檢查對應(yīng)的位圖上是否都標(biāo)記為1。
3. 應(yīng)用布隆過濾器
將實現(xiàn)的布隆過濾器應(yīng)用到Redis緩存層，從而避免數(shù)據(jù)庫查詢。具體代碼實現(xiàn)如下：
import redis
import math
class RedisBloomFilter:
def __init__(self, hosts, capacity, false_positive_prob, key):
self.r = redis.StrictRedis(hosts=hosts)
self.filter_key = key
self.bloom_filter = BloomFilter(capacity, false_positive_prob)
def add(self, key):
self.r.set(key, "1")
self.bloom_filter.add(key)
def contns(self, key):
if self.bloom_filter.__contns__(key):
if self.r.exists(key):
return True
return False
在具體實現(xiàn)中，添加一個新的key時，既可以將其寫入Redis，又可以將其添加到布隆過濾器中。當(dāng)檢查一個key是否存在時，先查詢布隆過濾器，如果已經(jīng)存在，則再檢查Redis緩存層，從而避免了數(shù)據(jù)庫查詢。
4. 驗證實驗結(jié)果
為了驗證以上實現(xiàn)的有效性，可以在一個模擬環(huán)境中進(jìn)行實驗。在實驗中，先使用模擬攻擊代碼不停地生成不存在于緩存中的key，然后根據(jù)不同的方案，比較查詢的時間和吞吐量。
具體代碼實現(xiàn)如下：
import time
import math
import redis
class CacheAside:
def __init__(self, hosts):
self.r = redis.StrictRedis(hosts=hosts)
def get(self, key):
val = self.r.get(key)
if not val:
val = self.get_from_db(key)
self.r.set(key, val)
return val
def get_from_db(self, key):
# simulate query time
time.sleep(0.1)
return key
class Limiter:
def __init__(self, hosts, max_requests, window_time):
self.r = redis.StrictRedis(hosts=hosts)
self.max_requests = max_requests
self.window_time = window_time
def is_allowed(self, key):
count = self.r.get(key)
if count and int(count) >= self.max_requests:
return False
else:
self.r.incr(key)
self.r.expire(key, self.window_time)
return True
def test_cache_aside():
cache_aside = CacheAside(['localhost'])
start_time = time.time()
for i in range(10000):
cache_aside.get(str(i))
end
四川成都云服務(wù)器租用托管【創(chuàng)新互聯(lián)】提供各地服務(wù)器租用，電信服務(wù)器托管、移動服務(wù)器托管、聯(lián)通服務(wù)器托管，云服務(wù)器虛擬主機(jī)租用。成都機(jī)房托管咨詢：13518219792
創(chuàng)新互聯(lián)（www.cdcxhl.com）擁有10多年的服務(wù)器租用、服務(wù)器托管、云服務(wù)器、虛擬主機(jī)、網(wǎng)站系統(tǒng)開發(fā)經(jīng)驗、開啟建站+互聯(lián)網(wǎng)銷售服務(wù)，與企業(yè)客戶共同成長，共創(chuàng)價值。
當(dāng)前題目：研究Redis穿透技術(shù)的有效性（redis穿透的有效性）
轉(zhuǎn)載源于：http://www.dlmjj.cn/article/cdppjco.html
