Add extra checks on hash lookups
This commit is contained in:
@@ -11,7 +11,14 @@ from core import r
|
||||
|
||||
|
||||
class SearchDenied:
|
||||
def __init__(self, value):
|
||||
def __init__(self, key, value):
|
||||
self.key = key
|
||||
self.value = value
|
||||
|
||||
|
||||
class LookupDenied:
|
||||
def __init__(self, key, value):
|
||||
self.key = key
|
||||
self.value = value
|
||||
|
||||
|
||||
@@ -84,7 +91,7 @@ def obfuscate_list(user, data):
|
||||
"""
|
||||
Obfuscate data in a list of dictionaries.
|
||||
"""
|
||||
if user.has_perm("core.bypass_obfuscation"):
|
||||
if user.has_perm("bypass_obfuscation"):
|
||||
return
|
||||
for index, item in enumerate(data):
|
||||
for key, value in item.items():
|
||||
@@ -123,7 +130,7 @@ def hash_list(user, data, hash_keys=False):
|
||||
"""
|
||||
Hash a list of dicts or a list with SipHash42.
|
||||
"""
|
||||
if user.has_perm("core.bypass_hashing"):
|
||||
if user.has_perm("bypass_hashing"):
|
||||
return
|
||||
cache = "cache.hash"
|
||||
hash_table = {}
|
||||
@@ -172,7 +179,12 @@ def hash_list(user, data, hash_keys=False):
|
||||
def hash_lookup(user, data_dict):
|
||||
cache = "cache.hash"
|
||||
hash_list = SortedSet()
|
||||
denied = []
|
||||
for key, value in list(data_dict.items()):
|
||||
if key in settings.SEARCH_FIELDS_DENY:
|
||||
if not user.has_perm("bypass_hashing"):
|
||||
data_dict[key] = SearchDenied(key=key, value=data_dict[key])
|
||||
denied.append(data_dict[key])
|
||||
if (
|
||||
key not in settings.WHITELIST_FIELDS
|
||||
and key not in settings.NO_OBFUSCATE_PARAMS
|
||||
@@ -193,8 +205,15 @@ def hash_lookup(user, data_dict):
|
||||
if not hashes:
|
||||
# Otherwise the user could inject plaintext search queries
|
||||
if not user.has_perm("bypass_hashing"):
|
||||
data_dict[key] = SearchDenied(value=data_dict[key])
|
||||
# del data_dict[key]
|
||||
data_dict[key] = SearchDenied(key=key, value=data_dict[key])
|
||||
denied.append(data_dict[key])
|
||||
continue
|
||||
else:
|
||||
# There are hashes here but there shouldn't be!
|
||||
if key in settings.TAG_SEARCH_DENY:
|
||||
data_dict[key] = LookupDenied(key=key, value=data_dict[key])
|
||||
denied.append(data_dict[key])
|
||||
continue
|
||||
|
||||
for hash in hashes:
|
||||
hash_list.add(hash)
|
||||
@@ -220,10 +239,11 @@ def hash_lookup(user, data_dict):
|
||||
for k2, v2 in data_dict[key].items():
|
||||
if hash in v2:
|
||||
data_dict[key][k2] = v2.replace(f"{hash}", total[hash])
|
||||
return denied
|
||||
|
||||
|
||||
def encrypt_list(user, data, secret):
|
||||
if user.has_perm("core.bypass_encryption"):
|
||||
if user.has_perm("bypass_encryption"):
|
||||
return
|
||||
cipher = Cipher(algorithms.AES(secret), ECB())
|
||||
for index, item in enumerate(data):
|
||||
|
||||
Reference in New Issue
Block a user