fix: validate cron expression before writing to crontab
This commit is contained in:
@@ -1,5 +1,5 @@
|
|||||||
import uuid
|
import uuid
|
||||||
from crontab import CronTab
|
from crontab import CronTab, CronSlices
|
||||||
|
|
||||||
|
|
||||||
class CrontabManager:
|
class CrontabManager:
|
||||||
@@ -39,6 +39,9 @@ class CrontabManager:
|
|||||||
command: str,
|
command: str,
|
||||||
is_enabled: bool = True,
|
is_enabled: bool = True,
|
||||||
):
|
):
|
||||||
|
if not CronSlices.is_valid(cron_expression):
|
||||||
|
raise ValueError(f"Invalid cron expression: {cron_expression!r}")
|
||||||
|
|
||||||
if not alarm_id:
|
if not alarm_id:
|
||||||
alarm_id = str(uuid.uuid4())
|
alarm_id = str(uuid.uuid4())
|
||||||
|
|
||||||
|
|||||||
@@ -164,6 +164,22 @@ def test_set_alarm_rejects_custom_command():
|
|||||||
assert "command" in str(res.json()["errors"])
|
assert "command" in str(res.json()["errors"])
|
||||||
|
|
||||||
|
|
||||||
|
def test_set_alarm_rejects_invalid_cron_expression():
|
||||||
|
"""Invalid cron expressions must be rejected at the API boundary."""
|
||||||
|
headers = {"X-API-Key": "test-secret"}
|
||||||
|
mutation = """
|
||||||
|
mutation {
|
||||||
|
setAlarm(cronExpression: "definitely not valid") {
|
||||||
|
id
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"""
|
||||||
|
res = client.post("/graphql", json={"query": mutation}, headers=headers)
|
||||||
|
assert res.status_code == 200
|
||||||
|
assert "errors" in res.json()
|
||||||
|
assert "Invalid cron expression" in str(res.json()["errors"])
|
||||||
|
|
||||||
|
|
||||||
def test_graphql_workflow():
|
def test_graphql_workflow():
|
||||||
headers = {"X-API-Key": "test-secret"}
|
headers = {"X-API-Key": "test-secret"}
|
||||||
|
|
||||||
|
|||||||
@@ -66,3 +66,9 @@ def test_delete_alarm(crontab_file):
|
|||||||
|
|
||||||
manager.delete_alarm(alarm_id)
|
manager.delete_alarm(alarm_id)
|
||||||
assert len(manager.get_alarms()) == 0
|
assert len(manager.get_alarms()) == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_set_alarm_rejects_invalid_cron_expression(crontab_file):
|
||||||
|
manager = CrontabManager(tabfile=crontab_file)
|
||||||
|
with pytest.raises(ValueError, match="Invalid cron expression"):
|
||||||
|
manager.set_alarm("test-id", "not-a-cron-expression", "cmd", True)
|
||||||
|
|||||||
Reference in New Issue
Block a user