mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2025-12-06 17:15:38 -08:00
195 lines
8.6 KiB
Python
Executable File
195 lines
8.6 KiB
Python
Executable File
import os
|
|
import json
|
|
|
|
import conf
|
|
from const import *
|
|
from logger import mylog, logResult
|
|
from helper import checkNewVersion
|
|
from utils.datetime_utils import timeNowDB, timeNow
|
|
|
|
# Register NetAlertX directories
|
|
INSTALL_PATH="/app"
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
# App state
|
|
#-------------------------------------------------------------------------------
|
|
# A class to manage the application state and to provide a frontend accessible API point
|
|
# To keep an existing value pass None
|
|
class app_state_class:
|
|
"""
|
|
Represents the current state of the application for frontend communication.
|
|
|
|
Attributes:
|
|
lastUpdated (str): Timestamp of the last update.
|
|
settingsSaved (int): Flag indicating if settings were saved.
|
|
settingsImported (int): Flag indicating if settings were imported.
|
|
showSpinner (bool): Whether the UI spinner should be shown.
|
|
processScan (bool): Whether a scan process is active.
|
|
graphQLServerStarted (int): Timestamp of GraphQL server start.
|
|
currentState (str): Current state string.
|
|
pluginsStates (dict): Per-plugin state information.
|
|
isNewVersion (bool): Flag indicating if a new version is available.
|
|
isNewVersionChecked (int): Timestamp of last version check.
|
|
"""
|
|
|
|
def __init__(self, currentState=None,
|
|
settingsSaved=None,
|
|
settingsImported=None,
|
|
showSpinner=None,
|
|
graphQLServerStarted=0,
|
|
processScan=False,
|
|
pluginsStates=None):
|
|
"""
|
|
Initialize the application state, optionally overwriting previous values.
|
|
|
|
Loads previous state from 'app_state.json' if available, otherwise initializes defaults.
|
|
New values provided via parameters overwrite previous state.
|
|
|
|
Args:
|
|
currentState (str, optional): Initial current state.
|
|
settingsSaved (int, optional): Initial settingsSaved flag.
|
|
settingsImported (int, optional): Initial settingsImported flag.
|
|
showSpinner (bool, optional): Initial showSpinner flag.
|
|
graphQLServerStarted (int, optional): Initial GraphQL server timestamp.
|
|
processScan (bool, optional): Initial processScan flag.
|
|
pluginsStates (dict, optional): Initial plugin states to merge with previous state.
|
|
"""
|
|
# json file containing the state to communicate with the frontend
|
|
stateFile = apiPath + 'app_state.json'
|
|
previousState = ""
|
|
|
|
# Update self
|
|
self.lastUpdated = str(timeNowDB())
|
|
|
|
if os.path.exists(stateFile):
|
|
try:
|
|
with open(stateFile, 'r') as json_file:
|
|
previousState = json.load(json_file)
|
|
except json.decoder.JSONDecodeError as e:
|
|
mylog('none', [f'[app_state_class] Failed to handle app_state.json: {e}'])
|
|
|
|
# Check if the file exists and recover previous values
|
|
if previousState != "":
|
|
self.settingsSaved = previousState.get("settingsSaved", 0)
|
|
self.settingsImported = previousState.get("settingsImported", 0)
|
|
self.processScan = previousState.get("processScan", False)
|
|
self.showSpinner = previousState.get("showSpinner", False)
|
|
self.isNewVersion = previousState.get("isNewVersion", False)
|
|
self.isNewVersionChecked = previousState.get("isNewVersionChecked", 0)
|
|
self.graphQLServerStarted = previousState.get("graphQLServerStarted", 0)
|
|
self.currentState = previousState.get("currentState", "Init")
|
|
self.pluginsStates = previousState.get("pluginsStates", {})
|
|
else: # init first time values
|
|
self.settingsSaved = 0
|
|
self.settingsImported = 0
|
|
self.showSpinner = False
|
|
self.processScan = False
|
|
self.isNewVersion = checkNewVersion()
|
|
self.isNewVersionChecked = int(timeNow().timestamp())
|
|
self.graphQLServerStarted = 0
|
|
self.currentState = "Init"
|
|
self.pluginsStates = {}
|
|
|
|
# Overwrite with provided parameters if supplied
|
|
if settingsSaved is not None:
|
|
self.settingsSaved = settingsSaved
|
|
if settingsImported is not None:
|
|
self.settingsImported = settingsImported
|
|
if showSpinner is not None:
|
|
self.showSpinner = showSpinner
|
|
if graphQLServerStarted is not None:
|
|
self.graphQLServerStarted = graphQLServerStarted
|
|
if processScan is not None:
|
|
self.processScan = processScan
|
|
if currentState is not None:
|
|
self.currentState = currentState
|
|
# Merge plugin states instead of overwriting
|
|
if pluginsStates is not None:
|
|
for plugin, state in pluginsStates.items():
|
|
if plugin in self.pluginsStates:
|
|
# Only update existing keys if both are dicts
|
|
if isinstance(self.pluginsStates[plugin], dict) and isinstance(state, dict):
|
|
self.pluginsStates[plugin].update(state)
|
|
else:
|
|
# Replace if types don't match
|
|
self.pluginsStates[plugin] = state
|
|
else:
|
|
# Optionally ignore or add new plugin entries
|
|
# To ignore new plugins, comment out the next line
|
|
self.pluginsStates[plugin] = state
|
|
|
|
# check for new version every hour and if currently not running new version
|
|
if self.isNewVersion is False and self.isNewVersionChecked + 3600 < int(timeNow().timestamp()):
|
|
self.isNewVersion = checkNewVersion()
|
|
self.isNewVersionChecked = int(timeNow().timestamp())
|
|
|
|
# Update .json file
|
|
# with open(stateFile, 'w') as json_file:
|
|
# json.dump(self, json_file, cls=AppStateEncoder, indent=4)
|
|
|
|
# Remove lastUpdated from the dictionary for comparison
|
|
currentStateDict = self.__dict__.copy()
|
|
currentStateDict.pop('lastUpdated', None)
|
|
|
|
# Compare current state with previous state before updating
|
|
if previousState != currentStateDict:
|
|
# Sanity check before saving the .json file
|
|
try:
|
|
json_data = json.dumps(self, cls=AppStateEncoder, indent=4)
|
|
with open(stateFile, 'w') as json_file:
|
|
json_file.write(json_data)
|
|
except (TypeError, ValueError) as e:
|
|
mylog('none', [f'[app_state_class] Failed to serialize object to JSON: {e}'])
|
|
|
|
return
|
|
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
# method to update the state
|
|
def updateState(newState = None,
|
|
settingsSaved = None,
|
|
settingsImported = None,
|
|
showSpinner = None,
|
|
graphQLServerStarted = None,
|
|
processScan = None,
|
|
pluginsStates=None):
|
|
"""
|
|
Convenience method to create or update the app state.
|
|
|
|
Args:
|
|
newState (str, optional): Current state to set.
|
|
settingsSaved (int, optional): Flag for settings saved.
|
|
settingsImported (int, optional): Flag for settings imported.
|
|
showSpinner (bool, optional): Flag to control UI spinner.
|
|
graphQLServerStarted (int, optional): Timestamp of GraphQL server start.
|
|
processScan (bool, optional): Flag indicating if a scan is active.
|
|
pluginsStates (dict, optional): Plugin state updates.
|
|
|
|
Returns:
|
|
app_state_class: Updated state object.
|
|
"""
|
|
return app_state_class( newState,
|
|
settingsSaved,
|
|
settingsImported,
|
|
showSpinner,
|
|
graphQLServerStarted,
|
|
processScan,
|
|
pluginsStates)
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
# Checks if the object has a __dict__ attribute. If it does, it assumes that it's an instance of a class and serializes its attributes dynamically.
|
|
class AppStateEncoder(json.JSONEncoder):
|
|
"""
|
|
JSON encoder for application state objects.
|
|
|
|
Automatically serializes objects with a __dict__ attribute.
|
|
"""
|
|
def default(self, obj):
|
|
if hasattr(obj, '__dict__'):
|
|
# If the object has a '__dict__', assume it's an instance of a class
|
|
return obj.__dict__
|
|
return super().default(obj)
|