strip leading/ending whitespace from task data.
This commit is contained in:
parent
fdec2356f9
commit
c367b7a5d3
3 changed files with 17 additions and 5 deletions
|
|
@ -18,7 +18,7 @@ from .encoder import Encoder
|
|||
from .Events import EventBus, Events, error, success
|
||||
from .Scheduler import Scheduler
|
||||
from .Singleton import Singleton
|
||||
from .Utils import init_class
|
||||
from .Utils import init_class, validate_url
|
||||
|
||||
LOG: logging.Logger = logging.getLogger("tasks")
|
||||
|
||||
|
|
@ -223,16 +223,26 @@ class Tasks(metaclass=Singleton):
|
|||
if not task.get("name"):
|
||||
msg = "No name found."
|
||||
raise ValueError(msg)
|
||||
else:
|
||||
task["name"] = task["name"].strip()
|
||||
|
||||
if not task.get("url"):
|
||||
msg = "No URL found."
|
||||
raise ValueError(msg)
|
||||
else:
|
||||
task["url"] = task["url"].strip()
|
||||
try:
|
||||
validate_url(task["url"], allow_internal=True)
|
||||
except ValueError as e:
|
||||
msg = f"Invalid URL format. '{e!s}'."
|
||||
raise ValueError(msg) from e
|
||||
|
||||
if task.get("timer"):
|
||||
try:
|
||||
from cronsim import CronSim
|
||||
|
||||
CronSim(task.get("timer"), datetime.now(UTC))
|
||||
task["timer"] = str(task["timer"]).strip()
|
||||
except Exception as e:
|
||||
msg = f"Invalid timer format. '{e!s}'."
|
||||
raise ValueError(msg) from e
|
||||
|
|
@ -242,6 +252,7 @@ class Tasks(metaclass=Singleton):
|
|||
from .Utils import arg_converter
|
||||
|
||||
arg_converter(args=task.get("cli"))
|
||||
task["cli"] = str(task["cli"]).strip()
|
||||
except Exception as e:
|
||||
msg = f"Invalid command options for yt-dlp. '{e!s}'."
|
||||
raise ValueError(msg) from e
|
||||
|
|
|
|||
|
|
@ -398,12 +398,13 @@ def is_private_address(hostname: str) -> bool:
|
|||
return True
|
||||
|
||||
|
||||
def validate_url(url: str) -> bool:
|
||||
def validate_url(url: str, allow_internal: bool = False) -> bool:
|
||||
"""
|
||||
Validate if the url is valid and allowed.
|
||||
|
||||
Args:
|
||||
url (str): URL to validate.
|
||||
allow_internal (bool): If True, allow internal URLs or private networks.
|
||||
|
||||
Returns:
|
||||
bool: True if the URL is valid and allowed.
|
||||
|
|
@ -429,8 +430,8 @@ def validate_url(url: str) -> bool:
|
|||
msg = "Invalid scheme usage. Only HTTP or HTTPS allowed."
|
||||
raise ValueError(msg)
|
||||
|
||||
hostname = parsed_url.host
|
||||
if not hostname or is_private_address(hostname):
|
||||
hostname: str | None = parsed_url.host
|
||||
if allow_internal is False and (not hostname or is_private_address(hostname)):
|
||||
msg = "Access to internal urls or private networks is not allowed."
|
||||
raise ValueError(msg)
|
||||
|
||||
|
|
|
|||
|
|
@ -75,7 +75,7 @@ async def tasks_add(request: Request, encoder: Encoder) -> Response:
|
|||
Tasks.validate(item)
|
||||
except ValueError as e:
|
||||
return web.json_response(
|
||||
{"error": f"Failed to validate task '{item.get('name')}'. '{e!s}'"},
|
||||
{"error": f"Failed to validate task '{item.get('name', '??')}'. '{e!s}'"},
|
||||
status=web.HTTPBadRequest.status_code,
|
||||
)
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue