diff --git a/routes/task/task_routes.py b/routes/task/task_routes.py index 49f65b313..0b7093d15 100644 --- a/routes/task/task_routes.py +++ b/routes/task/task_routes.py @@ -20,7 +20,7 @@ from src.task_action_policy import ( is_admin_only_task_action, owner_has_admin_task_privileges, ) -from src.task_scheduler import compute_next_run, HOUSEKEEPING_DEFAULTS +from src.task_scheduler import compute_next_run, is_valid_cron, HOUSEKEEPING_DEFAULTS from routes.prefs_routes import _load_for_user, _save_for_user logger = logging.getLogger(__name__) @@ -475,10 +475,7 @@ def setup_task_routes(task_scheduler) -> APIRouter: if req.trigger_type == "schedule" and req.schedule == "cron" and not req.cron_expression: raise HTTPException(400, "Cron expression is required for cron schedule") if req.trigger_type == "schedule" and req.schedule == "cron" and req.cron_expression: - try: - from croniter import croniter - croniter(req.cron_expression) - except Exception: + if not is_valid_cron(req.cron_expression): raise HTTPException(400, "Invalid cron expression") if req.trigger_type == "event" and not req.trigger_event: raise HTTPException(400, "Event name is required for event-triggered tasks") @@ -788,10 +785,7 @@ def setup_task_routes(task_scheduler) -> APIRouter: task.character_id = req.character_id or None if req.cron_expression is not None: if req.cron_expression: - try: - from croniter import croniter - croniter(req.cron_expression) - except Exception: + if not is_valid_cron(req.cron_expression): raise HTTPException(400, "Invalid cron expression") task.cron_expression = req.cron_expression or None diff --git a/src/task_scheduler.py b/src/task_scheduler.py index 50613ae72..23a573e36 100644 --- a/src/task_scheduler.py +++ b/src/task_scheduler.py @@ -123,6 +123,23 @@ async def _cached(key: Tuple, ttl: float, fetch: Callable[[], Awaitable[Any]]) - _shared_cache_pending.pop(key, None) +def is_valid_cron(cron_expression: str) -> bool: + """True for a standard 5-field cron expression that croniter can parse. + + croniter also accepts 6- and 7-field expressions with a seconds field, but + the scheduler works at one-minute resolution and would run such a task + every second. + """ + if len((cron_expression or "").split()) != 5: + return False + try: + from croniter import croniter + croniter(cron_expression) + except Exception: + return False + return True + + def compute_next_run(schedule: str, scheduled_time: str, scheduled_day: int = None, scheduled_date: datetime = None, @@ -166,6 +183,9 @@ def compute_next_run(schedule: str, scheduled_time: str, return dt.astimezone(timezone.utc).replace(tzinfo=None) if schedule == "cron" and cron_expression: + if not is_valid_cron(cron_expression): + logger.warning(f"Invalid cron expression '{cron_expression}'") + return None try: from croniter import croniter cron = croniter(cron_expression, now) diff --git a/tests/test_manage_tasks_cron.py b/tests/test_manage_tasks_cron.py index a3abe21d0..9738b7119 100644 --- a/tests/test_manage_tasks_cron.py +++ b/tests/test_manage_tasks_cron.py @@ -51,7 +51,7 @@ async def test_cron_create_edit_resume_and_invalid_edit_rollback(storage): @pytest.mark.asyncio -@pytest.mark.parametrize('expression', [None, '', 'bad', '99 99 * * *']) +@pytest.mark.parametrize('expression', [None, '', 'bad', '99 99 * * *', '* * * * * *', '* * * * * * *']) async def test_invalid_cron_does_not_create_inert_task(storage, expression): result = await call(action='create', name='Invalid', prompt='Test', schedule='cron', cron_expression=expression) @@ -171,3 +171,21 @@ async def test_invalid_cron_retime_rolls_back_all_edits(storage): row = db.get(database.ScheduledTask, result['task_id']) assert row.name == 'Digest' and row.cron_expression == '15 9 * * 1,3,5' assert row.next_run == datetime(2026, 10, 5, 9, 15) + + +@pytest.mark.parametrize('expression', ['* * * * *', '15 9 * * 1,3,5', '*/5 * * * *']) +def test_is_valid_cron_accepts_five_fields(expression): + from src.task_scheduler import is_valid_cron + assert is_valid_cron(expression) + + +@pytest.mark.parametrize('expression', [None, '', 'bad', '99 99 * * *', '* * * * * *', '* * * * * * *', '* * * *']) +def test_is_valid_cron_rejects_everything_else(expression): + from src.task_scheduler import is_valid_cron + assert not is_valid_cron(expression) + + +@pytest.mark.parametrize('expression', ['* * * * * *', '* * * * * * *']) +def test_compute_next_run_refuses_a_seconds_field(expression): + from src.task_scheduler import compute_next_run + assert compute_next_run('cron', '09:00', None, None, cron_expression=expression) is None