diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 9355daa7..48922a67 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -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 diff --git a/app/library/Utils.py b/app/library/Utils.py index a5182c78..43fa2c0b 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -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) diff --git a/app/routes/api/tasks.py b/app/routes/api/tasks.py index c326433b..92ecdea4 100644 --- a/app/routes/api/tasks.py +++ b/app/routes/api/tasks.py @@ -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, )