Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"Failed to read passwords file "
f"{self.settings[ConfigParameters.TABPY_PWD_FILE]}"
)
logger.critical(msg)
raise RuntimeError(msg)
else:
logger.info(
"Password file is not specified: " "Authentication is not enabled"
)
features = self._get_features()
self.settings[SettingsParameters.ApiVersions] = {"v1": {"features": features}}
set_parameter(
SettingsParameters.LogRequestContext,
ConfigParameters.TABPY_LOG_DETAILS,
default_val="false",
)
self.settings[SettingsParameters.LogRequestContext] = (
self.settings[SettingsParameters.LogRequestContext].lower() != "false"
)
call_context_state = (
"enabled"
if self.settings[SettingsParameters.LogRequestContext]
else "disabled"
)
logger.info(f"Call context logging is {call_context_state}")
set_parameter(
SettingsParameters.MaxRequestSizeInMb,
ConfigParameters.TABPY_MAX_REQUEST_SIZE_MB,
default_val=100,
def write_state_config(state, settings, logger=logging.getLogger(__name__)):
if SettingsParameters.StateFilePath in settings:
state_path = settings[SettingsParameters.StateFilePath]
else:
msg = f"{ConfigParameters.TABPY_STATE_PATH} is not set"
logger.log(logging.CRITICAL, msg)
raise ValueError(msg)
logger.log(logging.DEBUG, f"State path is {state_path}")
state_key = os.path.join(state_path, "state.ini")
tmp_state_file = state_key
with open(tmp_state_file, "w") as f:
state.write(f)
)
logger.debug(
f"File {state_file_path} not found, creating from "
f"template {state_file_template_path}..."
)
shutil.copy(state_file_template_path, state_file_path)
logger.info(f"Loading state from state file {state_file_path}")
tabpy_state = _get_state_from_file(state_file_dir)
self.tabpy_state = TabPyState(config=tabpy_state, settings=self.settings)
self.python_service = PythonServiceHandler(PythonService())
self.settings["compress_response"] = True
set_parameter(
SettingsParameters.StaticPath,
ConfigParameters.TABPY_STATIC_PATH,
default_val="./",
)
self.settings[SettingsParameters.StaticPath] = os.path.abspath(
self.settings[SettingsParameters.StaticPath]
)
logger.debug(
f"Static pages folder set to "
f'"{self.settings[SettingsParameters.StaticPath]}"'
)
# Set subdirectory from config if applicable
if tabpy_state.has_option("Service Info", "Subdirectory"):
self.subdirectory = "/" + tabpy_state.get("Service Info", "Subdirectory")
# If passwords file specified load credentials
set_parameter(ConfigParameters.TABPY_PWD_FILE, ConfigParameters.TABPY_PWD_FILE)
)
self.settings[SettingsParameters.EvaluateTimeout] = 30
pkg_path = os.path.dirname(tabpy.__file__)
set_parameter(
SettingsParameters.UploadDir,
ConfigParameters.TABPY_QUERY_OBJECT_PATH,
default_val=os.path.join(pkg_path, "tmp", "query_objects"),
)
if not os.path.exists(self.settings[SettingsParameters.UploadDir]):
os.makedirs(self.settings[SettingsParameters.UploadDir])
# set and validate transfer protocol
set_parameter(
SettingsParameters.TransferProtocol,
ConfigParameters.TABPY_TRANSFER_PROTOCOL,
default_val="http",
)
self.settings[SettingsParameters.TransferProtocol] = self.settings[
SettingsParameters.TransferProtocol
].lower()
set_parameter(
SettingsParameters.CertificateFile, ConfigParameters.TABPY_CERTIFICATE_FILE
)
set_parameter(SettingsParameters.KeyFile, ConfigParameters.TABPY_KEY_FILE)
self._validate_transfer_protocol_settings()
# if state.ini does not exist try and create it - remove
# last dependence on batch/shell script
set_parameter(
SettingsParameters.StateFilePath,
def _validate_cert_key_state(msg, cert_valid, key_valid):
cert_and_key_param = (
f"{ConfigParameters.TABPY_CERTIFICATE_FILE} and "
f"{ConfigParameters.TABPY_KEY_FILE}"
)
https_error = "Error using HTTPS: "
err = None
if not cert_valid and not key_valid:
err = https_error + msg.format(cert_and_key_param)
elif not cert_valid:
err = https_error + msg.format(ConfigParameters.TABPY_CERTIFICATE_FILE)
elif not key_valid:
err = https_error + msg.format(ConfigParameters.TABPY_KEY_FILE)
if err is not None:
logger.critical(err)
raise RuntimeError(err)
def _validate_cert_key_state(msg, cert_valid, key_valid):
cert_and_key_param = (
f"{ConfigParameters.TABPY_CERTIFICATE_FILE} and "
f"{ConfigParameters.TABPY_KEY_FILE}"
)
https_error = "Error using HTTPS: "
err = None
if not cert_valid and not key_valid:
err = https_error + msg.format(cert_and_key_param)
elif not cert_valid:
err = https_error + msg.format(ConfigParameters.TABPY_CERTIFICATE_FILE)
elif not key_valid:
err = https_error + msg.format(ConfigParameters.TABPY_KEY_FILE)
if err is not None:
logger.critical(err)
raise RuntimeError(err)
def _parse_pwd_file(self):
succeeded, self.credentials = parse_pwd_file(
self.settings[ConfigParameters.TABPY_PWD_FILE]
)
if succeeded and len(self.credentials) == 0:
logger.error("No credentials found")
succeeded = False
return succeeded
try:
self.settings[SettingsParameters.EvaluateTimeout] = float(
self.settings[SettingsParameters.EvaluateTimeout]
)
except ValueError:
logger.warning(
"Evaluate timeout must be a float type. Defaulting "
"to evaluate timeout of 30 seconds."
)
self.settings[SettingsParameters.EvaluateTimeout] = 30
pkg_path = os.path.dirname(tabpy.__file__)
set_parameter(
SettingsParameters.UploadDir,
ConfigParameters.TABPY_QUERY_OBJECT_PATH,
default_val=os.path.join(pkg_path, "tmp", "query_objects"),
)
if not os.path.exists(self.settings[SettingsParameters.UploadDir]):
os.makedirs(self.settings[SettingsParameters.UploadDir])
# set and validate transfer protocol
set_parameter(
SettingsParameters.TransferProtocol,
ConfigParameters.TABPY_TRANSFER_PROTOCOL,
default_val="http",
)
self.settings[SettingsParameters.TransferProtocol] = self.settings[
SettingsParameters.TransferProtocol
].lower()
set_parameter(
f"Parameter {settings_key} set to "
f'"{self.settings[settings_key]}" '
"from default value"
)
if not key_is_set:
logger.debug(f"Parameter {settings_key} is not set")
set_parameter(
SettingsParameters.Port, ConfigParameters.TABPY_PORT, default_val=9004
)
set_parameter(SettingsParameters.ServerVersion, None, default_val=__version__)
set_parameter(
SettingsParameters.EvaluateTimeout,
ConfigParameters.TABPY_EVALUATE_TIMEOUT,
default_val=30,
)
try:
self.settings[SettingsParameters.EvaluateTimeout] = float(
self.settings[SettingsParameters.EvaluateTimeout]
)
except ValueError:
logger.warning(
"Evaluate timeout must be a float type. Defaulting "
"to evaluate timeout of 30 seconds."
)
self.settings[SettingsParameters.EvaluateTimeout] = 30
pkg_path = os.path.dirname(tabpy.__file__)
set_parameter(