mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2025-12-07 09:36:05 -08:00
🔃 Sync Hub v0.6.31
This commit is contained in:
jokob-sk
@@ -14,7 +14,8 @@ sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
|
||||
from plugin_helper import Plugin_Object, Plugin_Objects, decodeBase64
|
||||
from plugin_utils import get_plugins_configs
|
||||
from logger import mylog
|
||||
from helper import timeNowTZ, get_setting_value, encrypt_data
|
||||
from helper import timeNowTZ, get_setting_value
|
||||
from cryptography import encrypt_data
|
||||
|
||||
# Define the current path and log file paths
|
||||
CUR_PATH = str(pathlib.Path(__file__).parent.resolve())
|
||||
|
||||
80
server/cryptography.py
Executable file
80
server/cryptography.py
Executable file
@@ -0,0 +1,80 @@
|
||||
# from cryptography.fernet import Fernet
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.Util.Padding import pad, unpad
|
||||
import base64
|
||||
import hashlib
|
||||
|
||||
|
||||
# FERET - Requires C compiler-------------------------------------------------------------------------
|
||||
|
||||
# def prepare_key(encryption_key):
|
||||
# if(len(encryption_key) < 32):
|
||||
# encryption_key = (int((32 / len(encryption_key)))+1 )*encryption_key
|
||||
|
||||
# key_bytearray = bytearray(encryption_key[:32], 'ASCII')
|
||||
|
||||
# return base64.urlsafe_b64encode(key_bytearray)
|
||||
|
||||
|
||||
# def encrypt_data(data, encryption_key):
|
||||
|
||||
# fernet = Fernet(prepare_key(encryption_key))
|
||||
|
||||
# # then use the Fernet class instance
|
||||
# # to encrypt the string string must
|
||||
# # be encoded to byte string before encryption
|
||||
# encrypted_data = fernet.encrypt(data.encode())
|
||||
# return encrypted_data
|
||||
|
||||
# def decrypt_data(data, encryption_key):
|
||||
|
||||
|
||||
# fernet = Fernet(prepare_key(encryption_key))
|
||||
|
||||
# # decrypt the encrypted string with the
|
||||
# # Fernet instance of the key,
|
||||
# # that was used for encrypting the string
|
||||
# # encoded byte string is returned by decrypt method,
|
||||
# # so decode it to string with decode methods
|
||||
# decrypted_data = fernet.decrypt(data).decode()
|
||||
# return decrypted_data
|
||||
|
||||
|
||||
# SIMPLE CRYPT - requeres C compiler -------------------------------------------------------------------------
|
||||
|
||||
# def prepare_key(encryption_key):
|
||||
# if len(encryption_key) < 32:
|
||||
# encryption_key = (encryption_key * ((32 // len(encryption_key)) + 1))[:32]
|
||||
# return encryption_key
|
||||
|
||||
# def encrypt_data(data, encryption_key):
|
||||
# key = prepare_key(encryption_key)
|
||||
# encrypted_data = encrypt(key, data)
|
||||
# return encrypted_data
|
||||
|
||||
# def decrypt_data(data, encryption_key):
|
||||
# key = prepare_key(encryption_key)
|
||||
# decrypted_data = decrypt(key, data).decode('utf-8')
|
||||
# return decrypted_data
|
||||
|
||||
# pycryptodome -------------------------------------------------------------------------
|
||||
|
||||
def prepare_key(encryption_key):
|
||||
key = hashlib.sha256(encryption_key.encode()).digest()
|
||||
return key
|
||||
|
||||
def encrypt_data(data, encryption_key):
|
||||
key = prepare_key(encryption_key)
|
||||
cipher = AES.new(key, AES.MODE_CBC)
|
||||
ct_bytes = cipher.encrypt(pad(data.encode('utf-8'), AES.block_size))
|
||||
iv = base64.b64encode(cipher.iv).decode('utf-8')
|
||||
ct = base64.b64encode(ct_bytes).decode('utf-8')
|
||||
return iv + ct
|
||||
|
||||
def decrypt_data(data, encryption_key):
|
||||
key = prepare_key(encryption_key)
|
||||
iv = base64.b64decode(data[:24])
|
||||
ct = base64.b64decode(data[24:])
|
||||
cipher = AES.new(key, AES.MODE_CBC, iv)
|
||||
pt = unpad(cipher.decrypt(ct), AES.block_size)
|
||||
return pt.decode('utf-8')
|
||||
@@ -266,17 +266,25 @@ def update_devices_data_from_scan (db):
|
||||
|
||||
# Update (unknown) or (name not found) Names if available
|
||||
mylog('debug','[Update Devices] - 4 Unknown Name')
|
||||
sql.execute ("""UPDATE Devices
|
||||
SET dev_NAME = (SELECT cur_Name FROM CurrentScan
|
||||
WHERE cur_MAC = dev_MAC
|
||||
and dev_Name not in ("(unknown)", "(name not found)", ""))
|
||||
WHERE (dev_Name in ("(unknown)", "(name not found)", "" )
|
||||
OR dev_Name IS NULL)
|
||||
AND EXISTS (SELECT 1 FROM CurrentScan
|
||||
WHERE cur_MAC = dev_MAC
|
||||
AND cur_Name IS NOT NULL
|
||||
AND cur_Name IS NOT 'null'
|
||||
AND cur_Name <> '') """)
|
||||
sql.execute (""" UPDATE Devices
|
||||
SET dev_NAME = COALESCE((
|
||||
SELECT cur_Name
|
||||
FROM CurrentScan
|
||||
WHERE cur_MAC = dev_MAC
|
||||
AND cur_Name IS NOT NULL
|
||||
AND cur_Name <> 'null'
|
||||
AND cur_Name <> ''
|
||||
), dev_NAME)
|
||||
WHERE (dev_NAME IN ('(unknown)', '(name not found)', '')
|
||||
OR dev_NAME IS NULL)
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM CurrentScan
|
||||
WHERE cur_MAC = dev_MAC
|
||||
AND cur_Name IS NOT NULL
|
||||
AND cur_Name <> 'null'
|
||||
AND cur_Name <> ''
|
||||
) """)
|
||||
|
||||
recordsToUpdate = []
|
||||
query = """SELECT * FROM Devices
|
||||
|
||||
@@ -15,8 +15,6 @@ from pathlib import Path
|
||||
import requests
|
||||
import base64
|
||||
import hashlib
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.Util.Padding import pad, unpad
|
||||
|
||||
|
||||
import conf
|
||||
@@ -802,85 +800,6 @@ def collect_lang_strings(json, pref, stringSqlParams):
|
||||
|
||||
return stringSqlParams
|
||||
|
||||
#-------------------------------------------------------------------------------
|
||||
# Cryptography
|
||||
#-------------------------------------------------------------------------------
|
||||
|
||||
|
||||
# FERET - Requires C compiler-------------------------------------------------------------------------
|
||||
|
||||
# def prepare_key(encryption_key):
|
||||
# if(len(encryption_key) < 32):
|
||||
# encryption_key = (int((32 / len(encryption_key)))+1 )*encryption_key
|
||||
|
||||
# key_bytearray = bytearray(encryption_key[:32], 'ASCII')
|
||||
|
||||
# return base64.urlsafe_b64encode(key_bytearray)
|
||||
|
||||
|
||||
# def encrypt_data(data, encryption_key):
|
||||
|
||||
# fernet = Fernet(prepare_key(encryption_key))
|
||||
|
||||
# # then use the Fernet class instance
|
||||
# # to encrypt the string string must
|
||||
# # be encoded to byte string before encryption
|
||||
# encrypted_data = fernet.encrypt(data.encode())
|
||||
# return encrypted_data
|
||||
|
||||
# def decrypt_data(data, encryption_key):
|
||||
|
||||
|
||||
# fernet = Fernet(prepare_key(encryption_key))
|
||||
|
||||
# # decrypt the encrypted string with the
|
||||
# # Fernet instance of the key,
|
||||
# # that was used for encrypting the string
|
||||
# # encoded byte string is returned by decrypt method,
|
||||
# # so decode it to string with decode methods
|
||||
# decrypted_data = fernet.decrypt(data).decode()
|
||||
# return decrypted_data
|
||||
|
||||
|
||||
# SIMPLE CRYPT - requeres C compiler -------------------------------------------------------------------------
|
||||
|
||||
# def prepare_key(encryption_key):
|
||||
# if len(encryption_key) < 32:
|
||||
# encryption_key = (encryption_key * ((32 // len(encryption_key)) + 1))[:32]
|
||||
# return encryption_key
|
||||
|
||||
# def encrypt_data(data, encryption_key):
|
||||
# key = prepare_key(encryption_key)
|
||||
# encrypted_data = encrypt(key, data)
|
||||
# return encrypted_data
|
||||
|
||||
# def decrypt_data(data, encryption_key):
|
||||
# key = prepare_key(encryption_key)
|
||||
# decrypted_data = decrypt(key, data).decode('utf-8')
|
||||
# return decrypted_data
|
||||
|
||||
# pycryptodome -------------------------------------------------------------------------
|
||||
|
||||
def prepare_key(encryption_key):
|
||||
key = hashlib.sha256(encryption_key.encode()).digest()
|
||||
return key
|
||||
|
||||
def encrypt_data(data, encryption_key):
|
||||
key = prepare_key(encryption_key)
|
||||
cipher = AES.new(key, AES.MODE_CBC)
|
||||
ct_bytes = cipher.encrypt(pad(data.encode('utf-8'), AES.block_size))
|
||||
iv = base64.b64encode(cipher.iv).decode('utf-8')
|
||||
ct = base64.b64encode(ct_bytes).decode('utf-8')
|
||||
return iv + ct
|
||||
|
||||
def decrypt_data(data, encryption_key):
|
||||
key = prepare_key(encryption_key)
|
||||
iv = base64.b64decode(data[:24])
|
||||
ct = base64.b64decode(data[24:])
|
||||
cipher = AES.new(key, AES.MODE_CBC, iv)
|
||||
pt = unpad(cipher.decrypt(ct), AES.block_size)
|
||||
return pt.decode('utf-8')
|
||||
|
||||
#-------------------------------------------------------------------------------
|
||||
# Misc
|
||||
#-------------------------------------------------------------------------------
|
||||
|
||||
@@ -11,10 +11,11 @@ from collections import namedtuple
|
||||
import conf
|
||||
from const import pluginsPath, logPath, applicationPath, reportTemplatesPath
|
||||
from logger import mylog
|
||||
from helper import timeNowTZ, updateState, get_file_content, write_file, get_setting, get_setting_value, decrypt_data
|
||||
from helper import timeNowTZ, updateState, get_file_content, write_file, get_setting, get_setting_value
|
||||
from api import update_api
|
||||
from plugin_utils import logEventStatusCounts, get_plugin_string, get_plugin_setting_obj, print_plugin_info, list_to_csv, combine_plugin_objects, resolve_wildcards_arr, handle_empty, custom_plugin_decoder
|
||||
from notification import Notification_obj
|
||||
from cryptography import decrypt_data
|
||||
|
||||
|
||||
#-------------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user