Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2025-10-30 22:36:21 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

ci(ruff): add bandit checks (#575)

Browse Source 
Added bandit checks to continuous integration.

Updated sources to pass bandit checks:
- replaced asserts
- added timeouts to requests
- added checks for process command execution
- changed to 127.0.0.1 as default IP address for EOS and EOSdash for security reasons

Added a rudimentary check for outdated config files.

BREAKING CHANGE: Default IP address for EOS and EOSdash changed to 127.0.0.1

Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com>
This commit is contained in:
Bobby Noelte
2025-06-03 08:30:37 +02:00
committed by GitHub
parent aa39ff475c
commit 3421b2303b
34 changed files with 163 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/akkudoktoreos/config/config.py
View File

@@ -376,6 +376,15 @@ class ConfigEOS(SingletonMixin, SettingsEOSDefaults):
def _setup(self, *args: Any, **kwargs: Any) -> None:
"""Re-initialize global settings."""
# Check for config file content/ version type
config_file, exists = self._get_config_file_path()
if exists:
with config_file.open("r", encoding="utf-8", newline=None) as f_config:
config_txt = f_config.read()
if '"directories": {' in config_txt or '"server_eos_host": ' in config_txt:
error_msg = f"Configuration file '{config_file}' is outdated. Please remove or update manually."
logger.error(error_msg)
raise ValueError(error_msg)
# Assure settings base knows EOS configuration
SettingsBaseModel.config = self
# (Re-)load settings
@@ -394,7 +403,9 @@ class ConfigEOS(SingletonMixin, SettingsEOSDefaults):
ValueError: If the `settings` is not a `SettingsEOS` instance.
"""
if not isinstance(settings, SettingsEOS):
raise ValueError(f"Settings must be an instance of SettingsEOS: '{settings}'.")
error_msg = f"Settings must be an instance of SettingsEOS: '{settings}'."
logger.error(error_msg)
raise ValueError(error_msg)
self.merge_settings_from_dict(settings.model_dump(exclude_none=True, exclude_unset=True))
@@ -471,10 +482,10 @@ class ConfigEOS(SingletonMixin, SettingsEOSDefaults):
@classmethod
def _get_config_file_path(cls) -> tuple[Path, bool]:
"""Finds the a valid configuration file or returns the desired path for a new config file.
"""Find a valid configuration file or return the desired path for a new config file.
Returns:
tuple[Path, bool]: The path to the configuration directory and if there is already a config file there
tuple[Path, bool]: The path to the configuration file and if there is already a config file there
"""
config_dirs = []
env_base_dir = os.getenv(cls.EOS_DIR)

2
src/akkudoktoreos/core/cache.py
View File

@@ -956,7 +956,7 @@ def cache_in_file(
logger.debug("Used cache file for function: " + func.__name__)
cache_file.seek(0)
if "b" in mode:
result = pickle.load(cache_file)
result = pickle.load(cache_file) # noqa: S301
else:
result = cache_file.read()
except Exception as e:

5
src/akkudoktoreos/core/decorators.py
View File

@@ -34,7 +34,7 @@ class classproperty:
argument and returns a value.
Raises:
AssertionError: If `fget` is not defined when `__get__` is called.
RuntimeError: If `fget` is not defined when `__get__` is called.
"""
def __init__(self, fget: Callable[[Any], Any]) -> None:
@@ -43,5 +43,6 @@ class classproperty:
def __get__(self, _: Any, owner_cls: Optional[type[Any]] = None) -> Any:
if owner_cls is None:
return self
assert self.fget is not None
if self.fget is None:
raise RuntimeError("'fget' not defined when `__get__` is called")
return self.fget(owner_cls)

3
src/akkudoktoreos/core/ems.py
View File

@@ -393,7 +393,8 @@ class EnergyManagement(SingletonMixin, ConfigMixin, PredictionMixin, PydanticBas
# Fetch objects
battery = self.battery
assert battery # to please mypy
if battery is None:
raise ValueError(f"battery not set: {battery}")
ev = self.ev
home_appliance = self.home_appliance
inverter = self.inverter

4
src/akkudoktoreos/core/pydantic.py
View File

@@ -450,8 +450,8 @@ class PydanticBaseModel(BaseModel, PydanticModelNestedValueMixin):
if expected_type is pendulum.DateTime or expected_type is AwareDatetime:
try:
value = to_datetime(value)
except:
pass
except Exception as e:
raise ValueError(f"Cannot convert {value!r} to datetime: {e}")
return value
# Override Pydantic’s serialization for all DateTime fields

3
src/akkudoktoreos/devices/battery.py
View File

@@ -121,7 +121,8 @@ class Battery(DeviceBase):
def _setup(self) -> None:
"""Sets up the battery parameters based on configuration or provided parameters."""
assert self.parameters is not None
if self.parameters is None:
raise ValueError(f"Parameters not set: {self.parameters}")
self.capacity_wh = self.parameters.capacity_wh
self.initial_soc_percentage = self.parameters.initial_soc_percentage
self.charging_efficiency = self.parameters.charging_efficiency

3
src/akkudoktoreos/devices/devicesabc.py
View File

@@ -170,7 +170,8 @@ class DevicesBase(DevicesStartEndMixin, PredictionMixin):
def add_device(self, device: Optional["DeviceBase"]) -> None:
if device is None:
return
assert device.device_id not in self.devices, f"{device.device_id} already registered"
if device.device_id in self.devices:
raise ValueError(f"{device.device_id} already registered")
self.devices[device.device_id] = device
def remove_device(self, device: Type["DeviceBase"] | str) -> bool:

3
src/akkudoktoreos/devices/generic.py
View File

@@ -34,7 +34,8 @@ class HomeAppliance(DeviceBase):
super().__init__(parameters)
def _setup(self) -> None:
assert self.parameters is not None
if self.parameters is None:
raise ValueError(f"Parameters not set: {self.parameters}")
self.load_curve = np.zeros(self.hours) # Initialize the load curve with zeros
self.duration_h = self.parameters.duration_h
self.consumption_wh = self.parameters.consumption_wh

6
src/akkudoktoreos/devices/inverter.py
View File

@@ -28,7 +28,8 @@ class Inverter(DeviceBase):
super().__init__(parameters)
def _setup(self) -> None:
assert self.parameters is not None
if self.parameters is None:
raise ValueError(f"Parameters not set: {self.parameters}")
if self.parameters.battery_id is None:
# For the moment raise exception
# TODO: Make battery configurable by config
@@ -41,7 +42,8 @@ class Inverter(DeviceBase):
) # Maximum power that the inverter can handle
def _post_setup(self) -> None:
assert self.parameters is not None
if self.parameters is None:
raise ValueError(f"Parameters not set: {self.parameters}")
self.battery = self.devices.get_device_by_id(self.parameters.battery_id)
def process_energy(

2
src/akkudoktoreos/optimization/genetic.py
View File

@@ -121,7 +121,7 @@ class optimization_problem(ConfigMixin, DevicesMixin, EnergyManagementSystemMixi
if self.fix_seed is not None:
random.seed(self.fix_seed)
elif logger.level == "DEBUG":
self.fix_seed = random.randint(1, 100000000000)
self.fix_seed = random.randint(1, 100000000000) # noqa: S311
random.seed(self.fix_seed)
def decode_charge_discharge(

13
src/akkudoktoreos/prediction/elecpriceakkudoktor.py
View File

@@ -104,12 +104,13 @@ class ElecPriceAkkudoktor(ElecPriceProvider):
- add the file cache again.
"""
source = "https://api.akkudoktor.net"
assert self.start_datetime # mypy fix
if not self.start_datetime:
raise ValueError(f"Start DateTime not set: {self.start_datetime}")
# Try to take data from 5 weeks back for prediction
date = to_datetime(self.start_datetime - to_duration("35 days"), as_string="YYYY-MM-DD")
last_date = to_datetime(self.end_datetime, as_string="YYYY-MM-DD")
url = f"{source}/prices?start={date}&end={last_date}&tz={self.config.general.timezone}"
response = requests.get(url)
response = requests.get(url, timeout=10)
logger.debug(f"Response from {url}: {response}")
response.raise_for_status() # Raise an error for bad responses
akkudoktor_data = self._validate_data(response.content)
@@ -148,7 +149,8 @@ class ElecPriceAkkudoktor(ElecPriceProvider):
"""
# Get Akkudoktor electricity price data
akkudoktor_data = self._request_forecast(force_update=force_update) # type: ignore
assert self.start_datetime # mypy fix
if not self.start_datetime:
raise ValueError(f"Start DateTime not set: {self.start_datetime}")
# Assumption that all lists are the same length and are ordered chronologically
# in ascending order and have the same timestamps.
@@ -178,7 +180,10 @@ class ElecPriceAkkudoktor(ElecPriceProvider):
)
amount_datasets = len(self.records)
assert highest_orig_datetime # mypy fix
if not highest_orig_datetime: # mypy fix
error_msg = f"Highest original datetime not available: {highest_orig_datetime}"
logger.error(error_msg)
raise ValueError(error_msg)
# some of our data is already in the future, so we need to predict less. If we got less data we increase the prediction hours
needed_hours = int(

2
src/akkudoktoreos/prediction/interpolator.py
View File

@@ -14,7 +14,7 @@ class SelfConsumptionProbabilityInterpolator:
self.filepath = filepath
# Load the RegularGridInterpolator
with open(self.filepath, "rb") as file:
self.interpolator: RegularGridInterpolator = pickle.load(file)
self.interpolator: RegularGridInterpolator = pickle.load(file) # noqa: S301
@lru_cache(maxsize=128)
def generate_points(

5
src/akkudoktoreos/prediction/pvforecastakkudoktor.py
View File

@@ -291,7 +291,7 @@ class PVForecastAkkudoktor(PVForecastProvider):
Raises:
ValueError: If the API response does not include expected `meta` data.
"""
response = requests.get(self._url())
response = requests.get(self._url(), timeout=10)
response.raise_for_status() # Raise an error for bad responses
logger.debug(f"Response from {self._url()}: {response}")
akkudoktor_data = self._validate_data(response.content)
@@ -332,7 +332,8 @@ class PVForecastAkkudoktor(PVForecastProvider):
logger.error(f"Akkudoktor schema change: {error_msg}")
raise ValueError(error_msg)
assert self.start_datetime # mypy fix
if not self.start_datetime:
raise ValueError(f"Start DateTime not set: {self.start_datetime}")
# Iterate over forecast data points
for forecast_values in zip(*akkudoktor_data.values):

7
src/akkudoktoreos/prediction/weatherbrightsky.py
View File

@@ -100,7 +100,8 @@ class WeatherBrightSky(WeatherProvider):
date = to_datetime(self.start_datetime, as_string=True)
last_date = to_datetime(self.end_datetime, as_string=True)
response = requests.get(
f"{source}/weather?lat={self.config.general.latitude}&lon={self.config.general.longitude}&date={date}&last_date={last_date}&tz={self.config.general.timezone}"
f"{source}/weather?lat={self.config.general.latitude}&lon={self.config.general.longitude}&date={date}&last_date={last_date}&tz={self.config.general.timezone}",
timeout=10,
)
response.raise_for_status() # Raise an error for bad responses
logger.debug(f"Response from {source}: {response}")
@@ -222,7 +223,7 @@ class WeatherBrightSky(WeatherProvider):
# Add Preciptable Water (PWAT) with a PVLib method.
key = WeatherDataRecord.key_from_description("Temperature (°C)")
assert key
assert key # noqa: S101
temperature = self.key_to_array(
key=key,
start_datetime=self.start_datetime,
@@ -235,7 +236,7 @@ class WeatherBrightSky(WeatherProvider):
logger.debug(debug_msg)
return
key = WeatherDataRecord.key_from_description("Relative Humidity (%)")
assert key
assert key # noqa: S101
humidity = self.key_to_array(
key=key,
start_datetime=self.start_datetime,

2
src/akkudoktoreos/prediction/weatherclearoutside.py
View File

@@ -93,7 +93,7 @@ class WeatherClearOutside(WeatherProvider):
source = "https://clearoutside.com/forecast"
latitude = round(self.config.general.latitude, 2)
longitude = round(self.config.general.longitude, 2)
response = requests.get(f"{source}/{latitude}/{longitude}?desktop=true")
response = requests.get(f"{source}/{latitude}/{longitude}?desktop=true", timeout=10)
response.raise_for_status() # Raise an error for bad responses
logger.debug(f"Response from {source}: {response}")
# We are working on fresh data (no cache), report update time

10
src/akkudoktoreos/server/dash/admin.py
View File

@@ -82,8 +82,8 @@ def AdminConfig(
try:
if config:
config_file_path = get_nested_value(config, ["general", "config_file_path"])
except:
pass
except Exception as e:
logger.debug(f"general.config_file_path: {e}")
# export config file
export_to_file_next_tag = to_datetime(as_string="YYYYMMDDHHmmss")
export_to_file_status = (None,)
@@ -95,7 +95,7 @@ def AdminConfig(
if data["action"] == "save_to_file":
# Safe current configuration to file
try:
result = requests.put(f"{server}/v1/config/file")
result = requests.put(f"{server}/v1/config/file", timeout=10)
result.raise_for_status()
config_file_path = result.json()["general"]["config_file_path"]
status = Success(f"Saved to '{config_file_path}' on '{eos_hostname}'")
@@ -143,7 +143,7 @@ def AdminConfig(
try:
with import_file_path.open("r", encoding="utf-8", newline=None) as fd:
import_config = json.load(fd)
result = requests.put(f"{server}/v1/config", json=import_config)
result = requests.put(f"{server}/v1/config", json=import_config, timeout=10)
result.raise_for_status()
import_from_file_status = Success(
f"Config imported from '{import_file_path}' on '{eosdash_hostname}'"
@@ -267,7 +267,7 @@ def Admin(eos_host: str, eos_port: Union[str, int], data: Optional[dict] = None)
# Get current configuration from server
server = f"http://{eos_host}:{eos_port}"
try:
result = requests.get(f"{server}/v1/config")
result = requests.get(f"{server}/v1/config", timeout=10)
result.raise_for_status()
config = result.json()
except requests.exceptions.HTTPError as e:

26
src/akkudoktoreos/server/dash/configuration.py
View File

@@ -218,7 +218,7 @@ def get_configuration(eos_host: str, eos_port: Union[str, int]) -> list[dict]:
# Get current configuration from server
try:
result = requests.get(f"{server}/v1/config")
result = requests.get(f"{server}/v1/config", timeout=10)
result.raise_for_status()
config = result.json()
except requests.exceptions.HTTPError as e:
@@ -303,9 +303,14 @@ def ConfigPlanesCard(
planes_update_open = True
plane_update_open = True
# Make mypy happy - should never trigger
assert isinstance(update_error, (str, type(None)))
assert isinstance(update_value, (str, type(None)))
assert isinstance(update_open, (bool, type(None)))
if (
not isinstance(update_error, (str, type(None)))
or not isinstance(update_value, (str, type(None)))
or not isinstance(update_open, (bool, type(None)))
):
error_msg = "update_error or update_value or update_open of wrong type."
logger.error(error_msg)
raise TypeError(error_msg)
plane_rows.append(
ConfigCard(
config["name"],
@@ -441,9 +446,14 @@ def Configuration(
update_value = config_update_latest.get(config["name"], {}).get("value")
update_open = config_update_latest.get(config["name"], {}).get("open")
# Make mypy happy - should never trigger
assert isinstance(update_error, (str, type(None)))
assert isinstance(update_value, (str, type(None)))
assert isinstance(update_open, (bool, type(None)))
if (
not isinstance(update_error, (str, type(None)))
or not isinstance(update_value, (str, type(None)))
or not isinstance(update_open, (bool, type(None)))
):
error_msg = "update_error or update_value or update_open of wrong type."
logger.error(error_msg)
raise TypeError(error_msg)
if (
config["type"]
== "Optional[list[akkudoktoreos.prediction.pvforecast.PVForecastPlaneSetting]]"
@@ -505,7 +515,7 @@ def ConfigKeyUpdate(eos_host: str, eos_port: Union[str, int], key: str, value: s
error = None
config = None
try:
response = requests.put(f"{server}/v1/config/{path}", json=data)
response = requests.put(f"{server}/v1/config/{path}", json=data, timeout=10)
response.raise_for_status()
config = response.json()
except requests.exceptions.HTTPError as err:

4
src/akkudoktoreos/server/dash/data/democonfig.json
View File

@@ -75,9 +75,9 @@
},
"server": {
"startup_eosdash": true,
"host": "0.0.0.0",
"host": "127.0.0.1",
"port": 8503,
"eosdash_host": "0.0.0.0",
"eosdash_host": "127.0.0.1",
"eosdash_port": 8504
},
"weather": {

14
src/akkudoktoreos/server/dash/demo.py
View File

@@ -186,7 +186,7 @@ def Demo(eos_host: str, eos_port: Union[str, int]) -> str:
# Get current configuration from server
try:
result = requests.get(f"{server}/v1/config")
result = requests.get(f"{server}/v1/config", timeout=10)
result.raise_for_status()
except requests.exceptions.HTTPError as err:
detail = result.json()["detail"]
@@ -200,12 +200,12 @@ def Demo(eos_host: str, eos_port: Union[str, int]) -> str:
with FILE_DEMOCONFIG.open("r", encoding="utf-8") as fd:
democonfig = json.load(fd)
try:
result = requests.put(f"{server}/v1/config", json=democonfig)
result = requests.put(f"{server}/v1/config", json=democonfig, timeout=10)
result.raise_for_status()
except requests.exceptions.HTTPError as err:
detail = result.json()["detail"]
# Try to reset to original config
requests.put(f"{server}/v1/config", json=config)
requests.put(f"{server}/v1/config", json=config, timeout=10)
return P(
f"Can not set demo configuration on {server}: {err}, {detail}",
cls="text-center",
@@ -213,12 +213,12 @@ def Demo(eos_host: str, eos_port: Union[str, int]) -> str:
# Update all predictions
try:
result = requests.post(f"{server}/v1/prediction/update")
result = requests.post(f"{server}/v1/prediction/update", timeout=10)
result.raise_for_status()
except requests.exceptions.HTTPError as err:
detail = result.json()["detail"]
# Try to reset to original config
requests.put(f"{server}/v1/config", json=config)
requests.put(f"{server}/v1/config", json=config, timeout=10)
return P(
f"Can not update predictions on {server}: {err}, {detail}",
cls="text-center",
@@ -239,7 +239,7 @@ def Demo(eos_host: str, eos_port: Union[str, int]) -> str:
"load_mean_adjusted",
],
}
result = requests.get(f"{server}/v1/prediction/dataframe", params=params)
result = requests.get(f"{server}/v1/prediction/dataframe", params=params, timeout=10)
result.raise_for_status()
predictions = PydanticDateTimeDataFrame(**result.json()).to_dataframe()
except requests.exceptions.HTTPError as err:
@@ -255,7 +255,7 @@ def Demo(eos_host: str, eos_port: Union[str, int]) -> str:
)
# Reset to original config
requests.put(f"{server}/v1/config", json=config)
requests.put(f"{server}/v1/config", json=config, timeout=10)
return Grid(
DemoPVForecast(predictions, democonfig),

2
src/akkudoktoreos/server/dash/footer.py
View File

@@ -24,7 +24,7 @@ def get_alive(eos_host: str, eos_port: Union[str, int]) -> str:
"""
result = requests.Response()
try:
result = requests.get(f"http://{eos_host}:{eos_port}/v1/health")
result = requests.get(f"http://{eos_host}:{eos_port}/v1/health", timeout=10)
if result.status_code == 200:
alive = result.json()["status"]
else:

25
src/akkudoktoreos/server/eos.py
View File

@@ -49,7 +49,11 @@ from akkudoktoreos.prediction.prediction import PredictionCommonSettings, get_pr
from akkudoktoreos.prediction.pvforecast import PVForecastCommonSettings
from akkudoktoreos.server.rest.error import create_error_page
from akkudoktoreos.server.rest.tasks import repeat_every
from akkudoktoreos.server.server import get_default_host, wait_for_port_free
from akkudoktoreos.server.server import (
get_default_host,
is_valid_ip_or_hostname,
wait_for_port_free,
)
from akkudoktoreos.utils.datetimeutil import to_datetime, to_duration
logger = get_logger(__name__)
@@ -100,6 +104,11 @@ def start_eosdash(
Raises:
RuntimeError: If the EOSdash server fails to start.
"""
if not is_valid_ip_or_hostname(host):
raise ValueError(f"Invalid EOSdash host: {host}")
if not is_valid_ip_or_hostname(eos_host):
raise ValueError(f"Invalid EOS host: {eos_host}")
eosdash_path = Path(__file__).parent.resolve().joinpath("eosdash.py")
# Do a one time check for port free to generate warnings if not so
@@ -130,7 +139,7 @@ def start_eosdash(
env["EOS_CONFIG_DIR"] = eos_config_dir
try:
server_process = subprocess.Popen(
server_process = subprocess.Popen( # noqa: S603
cmd,
env=env,
stdout=subprocess.PIPE,
@@ -240,10 +249,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
access_log = args.access_log
reload = args.reload
host = host if host else get_default_host()
port = port if port else 8504
eos_host = eos_host if eos_host else get_default_host()
eos_port = eos_port if eos_port else 8503
host = host if host else eos_host
port = port if port else 8504
eos_dir = str(config_eos.general.data_folder_path)
eos_config_dir = str(config_eos.general.config_folder_path)
@@ -370,7 +379,7 @@ async def fastapi_admin_server_restart_post() -> dict:
env["EOS_DIR"] = str(config_eos.general.data_folder_path)
env["EOS_CONFIG_DIR"] = str(config_eos.general.config_folder_path)
new_process = subprocess.Popen(
new_process = subprocess.Popen( # noqa: S603
[
sys.executable,
]
@@ -1208,7 +1217,7 @@ def redirect(request: Request, path: str) -> Union[HTMLResponse, RedirectRespons
if port is None:
port = 8504
# Make hostname Windows friendly
if host == "0.0.0.0" and os.name == "nt":
if host == "0.0.0.0" and os.name == "nt": # noqa: S104
host = "localhost"
url = f"http://{host}:{port}/"
error_page = create_error_page(
@@ -1225,7 +1234,7 @@ Did you want to connect to <a href="{url}" class="back-button">EOSdash</a>?
# Make hostname Windows friendly
host = str(config_eos.server.eosdash_host)
if host == "0.0.0.0" and os.name == "nt":
if host == "0.0.0.0" and os.name == "nt": # noqa: S104
host = "localhost"
if host and config_eos.server.eosdash_port:
# Redirect to EOSdash server
@@ -1258,7 +1267,7 @@ def run_eos(host: str, port: int, log_level: str, access_log: bool, reload: bool
None
"""
# Make hostname Windows friendly
if host == "0.0.0.0" and os.name == "nt":
if host == "0.0.0.0" and os.name == "nt": # noqa: S104
host = "localhost"
# Wait for EOS port to be free - e.g. in case of restart

2
src/akkudoktoreos/server/eosdash.py
View File

@@ -242,7 +242,7 @@ def run_eosdash() -> None:
reload = False
# Make hostname Windows friendly
if eosdash_host == "0.0.0.0" and os.name == "nt":
if eosdash_host == "0.0.0.0" and os.name == "nt": # noqa: S104
eosdash_host = "localhost"
# Wait for EOSdash port to be free - e.g. in case of restart

41
src/akkudoktoreos/server/server.py
View File

@@ -1,6 +1,7 @@
"""Server Module."""
import os
import ipaddress
import re
import time
from typing import Optional, Union
@@ -14,9 +15,39 @@ logger = get_logger(__name__)
def get_default_host() -> str:
if os.name == "nt":
return "127.0.0.1"
return "0.0.0.0"
"""Default host for EOS."""
return "127.0.0.1"
def is_valid_ip_or_hostname(value: str) -> bool:
"""Validate whether a string is a valid IP address (IPv4 or IPv6) or hostname.
This function first attempts to interpret the input as an IP address using the
standard library `ipaddress` module. If that fails, it checks whether the input
is a valid hostname according to RFC 1123, which allows domain names consisting
of alphanumeric characters and hyphens, with specific length and structure rules.
Args:
value (str): The input string to validate.
Returns:
bool: True if the input is a valid IP address or hostname, False otherwise.
"""
try:
ipaddress.ip_address(value)
return True
except ValueError:
pass
if len(value) > 253:
return False
hostname_regex = re.compile(
r"^(?=.{1,253}$)(?!-)[A-Z\d-]{1,63}(?<!-)"
r"(?:\.(?!-)[A-Z\d-]{1,63}(?<!-))*\.?$",
re.IGNORECASE,
)
return bool(hostname_regex.fullmatch(value))
def wait_for_port_free(port: int, timeout: int = 0, waiting_app_name: str = "App") -> bool:
@@ -110,6 +141,8 @@ class ServerCommonSettings(SettingsBaseModel):
cls, value: Optional[Union[str, IPvAnyAddress]]
) -> Optional[Union[str, IPvAnyAddress]]:
if isinstance(value, str):
if not is_valid_ip_or_hostname(value):
raise ValueError(f"Invalid host: {value}")
if value.lower() in ("localhost", "loopback"):
value = "127.0.0.1"
return value