percona · marcuscruz-percona · Aug 7, 2026 · Aug 7, 2026 · Aug 7, 2026 · Aug 7, 2026
@@ -162,11 +162,17 @@ class _TaskBackfillOutcome:
 
 
 def _task_write_from_task(task: Task, data: dict[str, Any]) -> TaskWrite:
-    """Build a ``TaskWrite`` envelope from an existing task row and ``data`` payload.
+    """Build the ``TaskWrite`` envelope that carries ``data`` through stamping.
+
+    The envelope is a carrier, not an update body: :func:`stamp_form_input` writes
+    only to ``write.data``, and that dict — not the envelope — is what the caller
+    persists. The two hook-path fields are therefore left unset rather than copied
+    off the row, so a stored path predating the ``TaskWrite`` allow-list cannot
+    fail validation here and cost an otherwise eligible row its backfill.
 
     :param task: The persisted task row.
     :param data: The ``data`` dict to carry on the write (including any stamp).
-    :return: A ``TaskWrite`` suitable for :meth:`~app.tasks.crud.TaskManager.update`.
+    :return: A ``TaskWrite`` carrying ``data`` for stamping.
     """
     return TaskWrite(
         name=task.name,
@@ -176,8 +182,6 @@ def _task_write_from_task(task: Task, data: dict[str, Any]) -> TaskWrite:
         is_template=task.is_template,
         protected=task.protected,
         alert_on_fail=task.alert_on_fail,
-        alert_detail_builder=task.alert_detail_builder,
-        run_result_recorder=task.run_result_recorder,
         output_files_path=task.output_files_path,
         anonymize_mask=task.anonymize_mask,
     )
@@ -292,6 +296,7 @@ def _reconstruct_validate_stamp(
 
     stamped_data = deepcopy(task.data)
     write = _task_write_from_task(task, stamped_data)
+
     try:
         stamp_form_input(write, validated_form)
     except Exception:

@@ -32,6 +32,7 @@
 from app.core.settings_override.registry import (
     hot_field,
     nested_overridable_field,
+    not_overridable_field,
 )
 from app.tasks.execution.executors.nomad import NomadExecutor
 
@@ -99,6 +100,14 @@ class TasksSettings(BaseYamlAppSettings):
     :param LOG_STREAM_EVICTION_MAX_ROWS: The maximum number of chunk rows the
         writer evicts per flush, bounding the per-append eviction work. Must be
         positive. Defaults to 1000.
+    :param HOOK_MODULE_ALLOWLIST: The module roots a per-task hook path
+        (``alert_detail_builder``, ``run_result_recorder``) may name. A path is
+        admitted when its module equals a root or is a submodule of one.
+        Environment- and YAML-only, and deliberately **not** exposed as a
+        runtime-overridable setting: widening the namespace live would itself be
+        a privilege-escalation path, since the resolved callable is imported and
+        invoked by the tasks service. Defaults to the namespace holding the
+        shipped task apps.
     """
 
     SETTINGS_PREFIXES: ClassVar[list[str]] = ["TASKS"]
@@ -124,6 +133,7 @@ class TasksSettings(BaseYamlAppSettings):
     )
     LOG_STREAM_CAP_BYTES: PositiveInt = hot_field(104857600, advanced=True)
     LOG_STREAM_EVICTION_MAX_ROWS: PositiveInt = hot_field(1000, advanced=True)
+    HOOK_MODULE_ALLOWLIST: tuple[str, ...] = not_overridable_field(("app.sep.apps",))
 
 
 tasks_settings: TasksSettings = OverridableSettingsProxy(

@@ -22,28 +22,104 @@
 resolve-and-cache boilerplate lives in one place.
 This process-local cache is correct because the import path deterministically
 resolves to the same callable in every worker and needs no invalidation.
+
+A hook path names a callable that this service imports and invokes, so the
+module it names is constrained to
+:attr:`app.tasks.config.TasksSettings.HOOK_MODULE_ALLOWLIST`. The same check
+runs at the write boundary (:class:`app.tasks.models.TaskWrite`) and here, so a
+path that reached the database by any other route fails closed at invoke time
+instead of being imported.
 """
 
 import importlib
+import logging
 from collections.abc import Callable
 from typing import Any
 
+logger = logging.getLogger(__name__)
+
 #: Cache of resolved callables, keyed by ``"module:function"`` path.
 _RESOLVED: dict[str, Callable[..., Any]] = {}
 
 
+class HookPathNotAllowedError(ValueError):
+    """Define exception for a hook path that is malformed or not allow-listed.
+
+    Subclasses :class:`ValueError` so the hook call sites, which already treat a
+    bad path as a skipped enrichment rather than a failure, keep degrading
+    gracefully.
+    """
+
+
+def _is_dotted_identifier(module_path: str) -> bool:
+    """Return whether every dot-separated segment is a Python identifier.
+
+    :param module_path: The module part of a hook path.
+    :return: ``True`` when the module part is a well-formed dotted name.
+    """
+    return bool(module_path) and all(
+        part.isidentifier() for part in module_path.split(".")
+    )
+
+
+def validate_hook_path(path: str, field: str = "hook path") -> str:
+    """Return ``path`` when it names an allow-listed callable, else raise.
+
+    A hook path is admitted only when it is a well-formed ``"module:function"``
+    pair naming a public function in a module under one of the configured
+    allow-listed roots. Everything else -- a malformed pair, a dunder or private
+    attribute, a module outside the namespace -- is rejected, because the
+    resolved callable is imported and invoked by the tasks service.
+
+    :param path: The candidate callable path in ``"module:function"`` form.
+    :param field: The name of the field carrying the path, quoted in the
+        rejection message. Defaults to a generic label.
+    :return: The validated path, unchanged.
+    :raises HookPathNotAllowedError: When the path is malformed or names a
+        module outside the allow-listed namespace.
+    """
+    # Deferred: app.tasks.config imports the Nomad executor, which imports
+    # app.tasks.models, which imports this module -- so tasks_settings does not
+    # exist yet while that chain is still initialising.
+    from app.tasks.config import tasks_settings
+
+    allowed = tuple(tasks_settings.HOOK_MODULE_ALLOWLIST)
+    module_path, _, func_name = path.partition(":")
+    if (
+        not _is_dotted_identifier(module_path)
+        or not func_name.isidentifier()
+        or func_name.startswith("_")
+    ):
+        reason = 'is not a "module:function" path naming a public callable'
+    elif not any(
+        module_path == root or module_path.startswith(f"{root}.") for root in allowed
+    ):
+        reason = "names a module outside the allow-listed namespace"
+    else:
+        return path
+
+    logger.warning("Rejected %s %r: it %s.", field, path, reason)
+    raise HookPathNotAllowedError(
+        f"{field} {path!r} {reason}; allow-listed module roots: {', '.join(allowed)}"
+    )
+
+
 def resolve_hook(path: str) -> Callable[..., Any]:
     """Import and return the callable named by a ``"module:function"`` path.
 
-    Cache the resolved callable so a repeated lookup for the same path skips the
-    import.
+    Validate the path against the allow-list before anything else, so neither a
+    denied module nor a cache entry poisoned before the allow-list narrowed can
+    be served. Cache the resolved callable so a repeated lookup for the same
+    path skips the import.
 
     :param path: The callable path in ``"module:function"`` form.
     :return: The resolved callable.
+    :raises HookPathNotAllowedError: When the path is malformed or names a
+        module outside the allow-listed namespace.
     :raises ImportError: When the named module cannot be imported.
     :raises AttributeError: When the module has no attribute ``function``.
-    :raises ValueError: When ``path`` carries no ``:`` separator.
     """
+    validate_hook_path(path)
     cached = _RESOLVED.get(path)
     if cached is not None:
         return cached

@@ -32,6 +32,7 @@
     field_validator,
     model_validator,
     ValidationError,
+    ValidationInfo,
 )
 from sqlalchemy import (
     BigInteger,
@@ -63,6 +64,7 @@
 from app.core.utils.path import resolve_payload_reference
 from app.tasks.anonymizer.config import anonymizer_settings
 from app.tasks.anonymizer.entities import PIIEntity
+from app.tasks.hook_resolver import validate_hook_path
 
 TASK_ALIAS_LENGTH = 100
 SYSTEM_USER = "SYSTEM"
@@ -532,6 +534,29 @@ class TaskWrite(TaskBase):
         logs and files generated by the task. Defaults to 0 (no anonymization).
     """
 
+    @field_validator("alert_detail_builder", "run_result_recorder")
+    @classmethod
+    def validate_hook_path_allow_listed(
+        cls, value: str | None, info: ValidationInfo
+    ) -> str | None:
+        """Reject a hook path outside the allow-listed module namespace.
+
+        A hook path is imported and invoked by the tasks service, so an
+        unconstrained value would let any caller who can write a task name an
+        arbitrary callable. The check lives here rather than on
+        :class:`TaskBase` so reading back a row whose stored path predates the
+        allow-list keeps working.
+
+        :param value: The candidate ``"module:function"`` path, or None.
+        :param info: The validation context, carrying the field name.
+        :return: The validated path, or None when the field is unset.
+        :raises HookPathNotAllowedError: When the path is malformed or names a
+            module outside the allow-listed namespace.
+        """
+        if value is None:
+            return None
+        return validate_hook_path(value, field=info.field_name or "hook path")
+
 
 class TaskExecuteRequest(BaseModel):
     """Represent a request to execute a task with additional metadata and payload.

@@ -0,0 +1 @@
+Task hook paths (`alert_detail_builder`, `run_result_recorder`) are now restricted to an allow-listed module namespace, closing a hole where any authenticated caller creating or updating a task could name an arbitrary importable callable for the tasks service to invoke.
@@ -258,6 +258,8 @@ default:
       NAME: tasks.db
     SYNC_LOCK_TTL: 300  # TaskHistory sync lock timeout (seconds)
     PRE_EXECUTION_CONNECTIVITY_CHECK: warn  # disabled | warn | block
+    HOOK_MODULE_ALLOWLIST:  # Module roots a task hook path may import from; extend for out-of-tree apps
+      - app.sep.apps
     NOMAD:
       ENDPOINT: http://127.0.0.1:4646
       SECURE: False

@@ -105,26 +105,42 @@ class GrafanaUserFactory(ModelFactory[GrafanaUser]):
 
 
 class TaskFactory(ModelFactory[Task]):
-    """Define factory for Task instances."""
+    """Define factory for Task instances.
+
+    Pins the hook-path fields to None so the random strings polyfactory would
+    otherwise generate for them do not trip the ``TaskWrite`` allow-list
+    validator wherever a built task is revalidated as a write.
+    """
 
     is_template: bool = False
     protected: bool = False
     backend: TaskBackendEnum = TaskBackendEnum.NOMAD
+    alert_detail_builder = None
+    run_result_recorder = None
 
 
 class PeriodicTaskFactory(SQLAlchemyFactory[PeriodicTask]):
     """Define factory for PeriodicTasks instances."""
 
 
 class GeneratedTaskFactory(ModelFactory[TaskWrite]):
-    """Define factory for GenerateTask instances."""
+    """Define factory for TaskWrite instances.
+
+    Pins the hook-path fields to None so factory-generated values do not trip
+    the ``TaskWrite`` allow-list validator.
+    """
+
+    alert_detail_builder = None
+    run_result_recorder = None
 
 
 class TaskResponseFactory(ModelFactory[TaskResponse]):
     """Define factory for TaskResponse instances.
 
     Pins ``backend`` to Nomad so the ``TaskBase`` proxy-backend validator (which
-    requires a ``data["task"]`` key) does not reject factory-generated data.
+    requires a ``data["task"]`` key) does not reject factory-generated data, and
+    the hook-path fields to None so they do not trip the ``TaskWrite``
+    allow-list validator wherever a built response is revalidated as a write.
     """
 
     backend: TaskBackendEnum = TaskBackendEnum.NOMAD
@@ -133,6 +149,8 @@ class TaskResponseFactory(ModelFactory[TaskResponse]):
     deleted_at = None
     created_by = None
     last_updated_by = None
+    alert_detail_builder = None
+    run_result_recorder = None
 
 
 class TaskHistoryResponseFactory(ModelFactory[TaskHistoryResponse]):

@@ -690,7 +690,7 @@ def test_create_threads_alert_detail_builder(
         """Assert the app's ``alert_detail_builder`` is stamped onto the posted task."""
         tasks_api = _make_tasks_api(created_task=_task_dict("new-task"))
         client = _client(
-            _synth_app(alert_detail_builder="pkg.mod:builder"),
+            _synth_app(alert_detail_builder="app.sep.apps.pkg.mod:builder"),
             tasks_api,
             regular_user,
             inventory_api=_make_inventory_api(),
@@ -705,15 +705,18 @@ def test_create_threads_alert_detail_builder(
         create_call = next(
             call for call in tasks_api.post.await_args_list if call.args[0] == "/"
         )
-        assert create_call.kwargs["json"]["alert_detail_builder"] == "pkg.mod:builder"
+        assert (
+            create_call.kwargs["json"]["alert_detail_builder"]
+            == "app.sep.apps.pkg.mod:builder"
+        )
 
     def test_create_threads_run_result_recorder(
         self, regular_user: CasdoorUser
     ) -> None:
         """Assert the app's ``run_result_recorder`` is stamped onto the posted task."""
         tasks_api = _make_tasks_api(created_task=_task_dict("new-task"))
         client = _client(
-            _synth_app(run_result_recorder="pkg.mod:recorder"),
+            _synth_app(run_result_recorder="app.sep.apps.pkg.mod:recorder"),
             tasks_api,
             regular_user,
             inventory_api=_make_inventory_api(),
@@ -728,7 +731,10 @@ def test_create_threads_run_result_recorder(
         create_call = next(
             call for call in tasks_api.post.await_args_list if call.args[0] == "/"
         )
-        assert create_call.kwargs["json"]["run_result_recorder"] == "pkg.mod:recorder"
+        assert (
+            create_call.kwargs["json"]["run_result_recorder"]
+            == "app.sep.apps.pkg.mod:recorder"
+        )
 
     def test_create_response_model_with_context_provider_succeeds(
         self, regular_user: CasdoorUser

@@ -259,6 +259,36 @@ def _invalid_body(_task: Task, _ctx: FormBackfillContext) -> dict:
     assert RESERVED_FORM_KEY not in task.data
 
 
+def test_backfill_single_task_stamps_a_row_whose_hook_path_predates_the_allow_list():
+    """A stored hook path the write model would reject still gets backfilled.
+
+    ``TaskWrite`` constrains hook paths to an allow-listed namespace, but the
+    envelope built here only carries ``data`` through stamping, so a row whose
+    stored path predates that constraint must not lose its backfill over it.
+    """
+    task = _minimal_task(data={"meta": {}})
+    task.alert_detail_builder = "app.sep.plugins.archives.alerts:build"
+
+    def _valid_body(_task: Task, _ctx: FormBackfillContext) -> dict:
+        return {
+            "task_name": _task.name,
+            "hostname": "executor-1",
+            "service_id": 1,
+            "recursion_method": "processlist",
+        }
+
+    entry = _entry(_valid_body)
+    ctx = FormBackfillContext(
+        log=logging.getLogger("test"), service_lookup=_EMPTY_SERVICE_LOOKUP
+    )
+
+    outcome = _backfill_single_task(task, entry, ctx)
+
+    assert outcome.label == "stamped"
+    assert outcome.stamped_data is not None
+    assert RESERVED_FORM_KEY in outcome.stamped_data
+
+
 @pytest.mark.asyncio
 async def test_backfill_single_task_stamp_preserves_audit_fields_on_persist(
     tasks_session: AsyncSession,

@@ -844,10 +844,10 @@ def test_alert_detail_builder_stamped_when_set(self) -> None:
             ResolvedEntities(service=service, entities={}),
             name="task-1",
             owner="ARCHIVER",
-            alert_detail_builder="pkg.mod:builder",
+            alert_detail_builder="app.sep.apps.pkg.mod:builder",
         )
 
-        assert write.alert_detail_builder == "pkg.mod:builder"
+        assert write.alert_detail_builder == "app.sep.apps.pkg.mod:builder"
 
     def test_alert_detail_builder_none_by_default(self) -> None:
         """Assert ``alert_detail_builder`` defaults to ``None`` when not supplied."""
@@ -887,10 +887,10 @@ def test_run_result_recorder_stamped_when_set(self) -> None:
             ResolvedEntities(service=service, entities={}),
             name="task-1",
             owner="BACKUPS",
-            run_result_recorder="pkg.mod:recorder",
+            run_result_recorder="app.sep.apps.pkg.mod:recorder",
         )
 
-        assert write.run_result_recorder == "pkg.mod:recorder"
+        assert write.run_result_recorder == "app.sep.apps.pkg.mod:recorder"
 
     def test_run_result_recorder_none_by_default(self) -> None:
         """Assert ``run_result_recorder`` defaults to ``None`` when not supplied."""
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		Task hook paths (`alert_detail_builder`, `run_result_recorder`) are now restricted to an allow-listed module namespace, closing a hole where any authenticated caller creating or updating a task could name an arbitrary importable callable for the tasks service to invoke.