fix(tasks): refuse cron expressions with a seconds field (#6824)

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
 

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)

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