diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3f8801419..ccd3db0a2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,6 +14,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
* Tag autocomplete suggestions are scoped to objects the current user can access
* Additional inline JavaScript values and activity-log rich-text previews are escaped or sanitized for their output context
* Added matching Django and Hasura validation for domain and static server names while preserving user access to create and manage shared inventory
+* Restricted Django Q scheduled tasks to a server-controlled allowlist (Closes #911)
+ * The admin panel now exposes only approved functions, Slack notification hooks, and validated task arguments
+ * Optional system commands must be configured as fixed argument vectors on the server and run without a shell
+ * Schedule saves, queue submissions, result hooks, and worker execution enforce the policy independently
+ * Added `audit_django_q_policy` to identify or pause existing schedules that do not satisfy the configured policy
## [7.2.4] - 21 July 2026
diff --git a/DOCS/development/modifying-code.mdx b/DOCS/development/modifying-code.mdx
index 93306a654..85047d084 100644
--- a/DOCS/development/modifying-code.mdx
+++ b/DOCS/development/modifying-code.mdx
@@ -7,7 +7,7 @@ description: "Modifying and then deploying code changes"
While running Ghostwriter using `local.yml`, you can make changes on the fly without needing to run any Docker commands. Django smartly restarts the server when it detects a file change. Also, the local deployment mounts Ghostwriter's directory on your host, so changes to the code (Python or templates) will be live once the file is saved.
-The `build, stop, rm, up -d` commands need to be run only if you add a new library or if you add a new task to **tasks.py** **(only after initially adding the new function, not each time you change the function).
+The `build, stop, rm, up -d` commands need to be run only if you add a new library or if you add a new task to **tasks.py** **(only after initially adding the new function, not each time you change the function).** A new queued task must also be added to `GHOSTWRITER_DJANGO_Q_INTERNAL_TASKS` or `GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS`, as appropriate.
`docker-compose -f local.yml stop; docker-compose -f local.yml rm -f; docker-compose -f local.yml build; docker-compose -f local.yml up -d`
diff --git a/DOCS/features/access-authentication-and-session-controls/session-management.mdx b/DOCS/features/access-authentication-and-session-controls/session-management.mdx
index 9ad8519ea..7d870ba4d 100644
--- a/DOCS/features/access-authentication-and-session-controls/session-management.mdx
+++ b/DOCS/features/access-authentication-and-session-controls/session-management.mdx
@@ -39,7 +39,7 @@ Ghostwriter also tracks JWT sessions created by the GraphQL `login` mutation. Th
It is recommended you clear expired sessions on a regular basis to keep the session tables tidy. Use the `clear_expired_sessions` management command for this cleanup. The command wraps Django's built-in `clearsessions` command and also deletes expired GraphQL login sessions from Ghostwriter's `UserSession` table.
-For a scheduled task, call `django.core.management.call_command` and pass `"clear_expired_sessions"` as its only argument. Set up a task like this one that runs daily with the cron scheduler. For example, `0 5 * * *` will run it every day at 5:00 AM.
+For a scheduled task, select the allowlisted `ghostwriter.home.django_q_tasks.clear_expired_sessions` function. The wrapper accepts no arguments and runs only the `clear_expired_sessions` management command. Set up a task like this one with the cron scheduler. For example, `0 5 * * *` will run it every day at 5:00 AM.
diff --git a/DOCS/features/background-tasks.mdx b/DOCS/features/background-tasks.mdx
index 0099420b4..de87a03be 100644
--- a/DOCS/features/background-tasks.mdx
+++ b/DOCS/features/background-tasks.mdx
@@ -5,7 +5,7 @@ description: Configuring and scheduling background tasks
Ghostwriter uses the [Django Q project](https://django-q.readthedocs.io/en/latest/) for queuing and managing background tasks. Django Q hands off tasks to the Redis server (already installed and running in Docker).
-Tasks are defined in the `tasks.py` file. These tasks can be executed on demand or on a schedule.
+Tasks are Python functions, usually defined in a `tasks.py` module. Ghostwriter executes some tasks on demand from the application. Only functions approved by the server-side Django Q policy can be added to a schedule in the admin panel.
Tasks can be queued in a few different ways:
@@ -14,4 +14,3 @@ Tasks can be queued in a few different ways:
* Use the buttons (various) in Ghostwriter's web interface.
* Use a REST API endpoint (not yet available).
-
diff --git a/DOCS/features/background-tasks/prebuilt-tasks.mdx b/DOCS/features/background-tasks/prebuilt-tasks.mdx
index ccde4a9e9..52aa46353 100644
--- a/DOCS/features/background-tasks/prebuilt-tasks.mdx
+++ b/DOCS/features/background-tasks/prebuilt-tasks.mdx
@@ -9,7 +9,7 @@ The following information is for Ghostwriter's provided tasks
The `ghostwriter.shepherd.tasks.release_domains` function checks if the currently checked-out domain names are due to be released. If Slack is enabled, it sends a Slack message when the domain's release date is tomorrow or today. If the release date is today, the domain is also released back into the pool.
-The task accepts an optional `reset_dns` argument that defaults to `False`. If this argument is set to `True`, the task will use the Namecheap API to reset the domain's DNS records upon release. This feature requires:
+The task accepts an optional `no_action` argument that defaults to `False`. Set it to `True` to preview the domains due for release without changing their availability. Domains configured to reset their DNS records will use the Namecheap API during a normal release. DNS reset requires:
* Namecheap API enabled and configured
@@ -43,25 +43,24 @@ If you will be using this task, the Q cluster needs to be started using an admin
## Namecheap Synchronization
-The `tasks.fetch_namecheap_domains` uses the Namecheap API to fetch all domains from the registrar for the configured account and synchronizes that information with the domain library. If a domain name is found in the library already, the task will update that record.
+The `ghostwriter.shepherd.tasks.fetch_namecheap_domains` task uses the Namecheap API to fetch all domains from the registrar for the configured account and synchronizes that information with the domain library. If a domain name is found in the library already, the task will update that record.
If a Namecheap-registered domain in the Ghostwriter library is not found in the fresh list of domains, the task will mark that domain as expired.
## Cloud Monitoring
-The `review_cloud_infrastructure` task uses the Amazon Web Services and Digital Ocean APIs to check all running server instances and compare those to cloud servers attached to projects. If a project is marked as complete, the task will flag any running cloud servers attached to that project for review.
+The `ghostwriter.shepherd.tasks.review_cloud_infrastructure` task uses the Amazon Web Services and Digital Ocean APIs to check all running server instances and compare those to cloud servers attached to projects. If a project is marked as complete, the task will flag any running cloud servers attached to that project for review.
The task will send a Slack message with all relevant server data to the project's channel if Slack is enabled.
## Update Expiration
-The `check_expiration` task checks the expiration dates of all domains in the domain library and compares them to the current date. If the domain is set to auto-renew, the task will increment the expiration date by one year. Otherwise, the task will mark the domain as expired.
+The `ghostwriter.shepherd.tasks.check_expiration` task checks the expiration dates of all domains in the domain library and compares them to the current date. If the domain is set to auto-renew, the task will increment the expiration date by one year. Otherwise, the task will mark the domain as expired.
## Operation Log Monitor
-The `ghostwriter.modules.oplog_monitors.review_activity_logs` task checks operation logs for activity. It reviews all logs for all active projects. By default, the task looks for an entry created in the past 24 hours. You can change the hours by setting the `hours` argument when scheduling the task.
+The `ghostwriter.modules.oplog_monitors.review_active_logs` task checks operation logs for activity. It reviews all logs for all active projects. By default, the task looks for an entry created in the past 24 hours. You can change the hours by setting the `hours` argument when scheduling the task.
The task outputs JSON detailing all activity logs that appear to be idle. This is useful for catching automated activity logging that may have started failing.
If a Slack webhook is enabled and configured, the task will also send a Slack message to the project's Slack channel (if configured) or the global Slack channel.
-
diff --git a/DOCS/features/background-tasks/scheduled-tasks.mdx b/DOCS/features/background-tasks/scheduled-tasks.mdx
index 143fa4fb4..91b82adeb 100644
--- a/DOCS/features/background-tasks/scheduled-tasks.mdx
+++ b/DOCS/features/background-tasks/scheduled-tasks.mdx
@@ -6,15 +6,127 @@ description: "Configuring tasks to run repeatedly on a schedule"
## Scheduling a Task
-Tasks are scheduled in the Django admin panel under _Django Q_ and _Scheduled Tasks_. Add a new task, give the task a name, and tell it which function to run. Functions must be prefixed with `tasks.` to tell Django Q to pull the functions from `tasks.py`. The final requirement is a schedule. Tasks can be run once or repeatedly (e.g. minutes, hourly, daily, weekly).
+Tasks are scheduled in the Django admin panel under _Django Q_ and _Scheduled Tasks_. Add a new task, give the task a name, select one of the server-approved
+functions, and configure its schedule. Tasks can run once or repeatedly (e.g. minutes, hourly, daily, weekly). You can schedule them by time or using `cron`.
-To schedule the `release_servers()` function inside `tasks.py`, you would enter the function name as: `tasks.release_servers`
+The function and hook lists are controlled by a server-side allowlist. An administrator who only has access to the admin panel cannot add Python functions
+or system commands to these lists.
-You can provide arguments for the functions as needed. Ghostwriter's scheduled tasks do not require arguments, but some have optional arguments. For example, the `scan_servers()` function accepts an `active_only` argument to restrict scanning only to servers that have been checked out for a project.
+
+You can provide approved arguments for functions that accept them. Ghostwriter rejects unknown arguments and values with the wrong type. For example,
+`ghostwriter.shepherd.tasks.scan_servers` accepts `only_active=True` to restrict scanning to servers that are in use.
### Example Scheduled Task
-Visit the Django Q database from the admin panel to access the _Scheduled Tasks_. You may wish to create a scheduled task to automatically release domains at the end of a project. There is a task for this already in `tasks.py`, `ghostwriter.shepherd.tasks.release_domains`.
+Visit the Django Q database from the admin panel to access the _Scheduled Tasks_. For example, you may wish to create a scheduled task to automatically release domains
+at the end of a project. There is a task for this already in `tasks.py`, `ghostwriter.shepherd.tasks.release_domains`. It appears in the dropdown as _Release Domains_.
+
+### Available Task Arguments
+
+Arguments may be supplied positionally in the _Arguments_ field or by name in
+the _Keyword arguments_ field. Both fields use Python literal syntax, so string
+values must be quoted. For example, enter `'nightly-backup'` in _Arguments_ or
+`command_name='nightly-backup'` in _Keyword arguments_. Do not provide the same
+parameter in both fields.
+
+| Function | Available parameters |
+| --- | --- |
+| `ghostwriter.reporting.tasks.archive_projects` | None |
+| `ghostwriter.rolodex.tasks.check_project_freshness` | None |
+| `ghostwriter.shepherd.tasks.check_domains` | `domain_id`: positive integer or `None`; omit it to check every domain |
+| `ghostwriter.shepherd.tasks.check_expiration` | None |
+| `ghostwriter.shepherd.tasks.fetch_namecheap_domains` | None |
+| `ghostwriter.shepherd.tasks.release_domains` | `no_action`: boolean; `True` previews without releasing |
+| `ghostwriter.shepherd.tasks.release_servers` | `no_action`: boolean; `True` previews without releasing |
+| `ghostwriter.shepherd.tasks.review_cloud_infrastructure` | `aws_only_running`: boolean; `do_only_running`: boolean |
+| `ghostwriter.shepherd.tasks.scan_servers` | `only_active`: boolean; `True` scans only servers in use |
+| `ghostwriter.shepherd.tasks.update_dns` | `domain`: positive integer or `None`; omit it to update every domain |
+| `ghostwriter.modules.oplog_monitors.review_active_logs` | `hours`: integer from 1 through 8,760; defaults to 24 |
+| `ghostwriter.home.django_q_tasks.clear_expired_sessions` | None |
+| Fixed system-command runner | `command_name`: one of the names in `GHOSTWRITER_DJANGO_Q_COMMANDS` |
+
+## Configuring the Server Allowlist
+
+Ghostwriter includes its prebuilt scheduled tasks in the default policy. Server operators can replace or extend `GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS` from a settings
+fragment such as `settings/10-django-q-policy.py`. The Ghostwriter CLI mounts the `settings` directory read-only into the web and queue containers.
+
+```python
+GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS = {
+ **GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS,
+ "organization.tasks.refresh_inventory": {
+ "label": "Refresh Organization Inventory",
+ "args": [],
+ "kwargs": {
+ "active_only": {"type": "bool"},
+ },
+ },
+}
+```
+
+Task entries use exact dotted callable paths. Prefixes and wildcards are not supported. Argument policies support `bool`, `float`, `int`, and `str` values, along with
+`name`, `required`, `required_parameters`, `nullable`, `choices`, `min`, and `max` restrictions. A named positional argument can also appear in `kwargs`, allowing
+administrators to use either input style. Set `allow_any_arguments` only for a trusted function that performs its own strict input validation.
+
+Result hooks use the separate `GHOSTWRITER_DJANGO_Q_SCHEDULE_HOOKS` mapping. Ghostwriter exposes its built-in `ghostwriter.modules.notifications_slack.send_slack_complete_msg`
+hook by default. Generic execution functions such as `os.system`, `subprocess.run`, and `django.core.management.call_command` should never be allowlisted.
+
+### Fixed System Commands
+
+Server operators can expose a fixed command by adding it to `GHOSTWRITER_DJANGO_Q_COMMANDS`:
+
+```python
+GHOSTWRITER_DJANGO_Q_COMMANDS = {
+ "nightly-backup": {
+ "argv": ["/usr/local/bin/ghostwriter-backup", "--quiet"],
+ "timeout": 1800,
+ "cwd": "/app",
+ "env": {},
+ },
+}
+```
+
+The executable path and all arguments are fixed in the server configuration. Commands run without a shell and with only the explicitly configured environment. The admin
+panel permits selecting the command name but does not permit editing the executable, arguments, working directory, or environment.
+
+Restart both the Ghostwriter web and queue services after changing the policy. Changes are intentionally not loaded from the database or admin panel.
+
+### Auditing Existing Schedules
+
+The task allowlist is deployment-specific, so Ghostwriter does not use a Django
+data migration to modify existing schedules. Instead, use the policy audit
+command as an operational migration when upgrading or enabling a more
+restrictive policy.
+
+For the safest rollout:
+
+1. Stop the Django Q cluster so an old scheduler or worker cannot execute tasks
+ during the upgrade.
+2. Deploy the new Ghostwriter code and the intended server-side allowlist.
+3. Audit the existing schedules:
+
+```sh
+python manage.py audit_django_q_policy --check
+```
+
+4. Add any legitimate custom tasks reported by the audit to the server-side
+ allowlist, or pause every schedule that does not satisfy the current policy:
+
+```sh
+python manage.py audit_django_q_policy --pause-disallowed
+```
+
+5. Restart the Ghostwriter web and Django Q services.
+
+Disallowed schedules are paused by setting `repeats=0`; they are not deleted.
+This preserves their configuration for later review. If a disallowed schedule
+is missed by the audit, the restricted scheduler pauses it when it becomes due.
+A task already present in the queue is checked again by the worker and recorded
+as a failed task without importing or executing the denied function. Historical
+successful and failed task records remain available, but the admin panel cannot
+resubmit them unless they satisfy the current policy.
-This example executes `ghostwriter.shepherd.tasks.release_domains` under the name `Domain Release` every day at 15:39 UTC with no additional args or kwargs.
+
+The allowlist limits administrators who only control the application or Django admin panel. Anyone who can modify the server configuration, application code, queue,
+database, or application secrets is already inside the server trust boundary.
+
diff --git a/config/settings/base.py b/config/settings/base.py
index 6240bde31..ec619f7bc 100644
--- a/config/settings/base.py
+++ b/config/settings/base.py
@@ -425,6 +425,132 @@ def env_float(name, default):
"redis": env("QCLUSTER_CONNECTION", default={"host": "redis", "port": 6379, "db": 0}),
}
+# Only tasks in this server-side policy can be created through the Django Q
+# schedule admin. Deployments can replace or extend this dictionary from a
+# settings fragment in ``production.d``. Argument specifications are enforced
+# when a schedule is saved, when it is enqueued, and again by the queue worker.
+GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS = {
+ "ghostwriter.reporting.tasks.archive_projects": {
+ "label": "Archive Completed Projects",
+ "args": [],
+ "kwargs": {},
+ },
+ "ghostwriter.rolodex.tasks.check_project_freshness": {
+ "label": "Check Project Freshness",
+ "args": [],
+ "kwargs": {},
+ },
+ "ghostwriter.shepherd.tasks.check_domains": {
+ "label": "Check Domain Categorization",
+ "args": [
+ {
+ "name": "domain_id",
+ "type": "int",
+ "nullable": True,
+ "min": 1,
+ "required": False,
+ }
+ ],
+ "kwargs": {"domain_id": {"type": "int", "nullable": True, "min": 1}},
+ },
+ "ghostwriter.shepherd.tasks.check_expiration": {
+ "label": "Check Domain Expiration",
+ "args": [],
+ "kwargs": {},
+ },
+ "ghostwriter.shepherd.tasks.fetch_namecheap_domains": {
+ "label": "Synchronize Namecheap Domains",
+ "args": [],
+ "kwargs": {},
+ },
+ "ghostwriter.shepherd.tasks.release_domains": {
+ "label": "Release Domains",
+ "args": [{"name": "no_action", "type": "bool", "required": False}],
+ "kwargs": {"no_action": {"type": "bool"}},
+ },
+ "ghostwriter.shepherd.tasks.release_servers": {
+ "label": "Release Servers",
+ "args": [{"name": "no_action", "type": "bool", "required": False}],
+ "kwargs": {"no_action": {"type": "bool"}},
+ },
+ "ghostwriter.shepherd.tasks.review_cloud_infrastructure": {
+ "label": "Review Cloud Infrastructure",
+ "args": [
+ {"name": "aws_only_running", "type": "bool", "required": False},
+ {"name": "do_only_running", "type": "bool", "required": False},
+ ],
+ "kwargs": {
+ "aws_only_running": {"type": "bool"},
+ "do_only_running": {"type": "bool"},
+ },
+ },
+ "ghostwriter.shepherd.tasks.scan_servers": {
+ "label": "Scan Servers",
+ "args": [{"name": "only_active", "type": "bool", "required": False}],
+ "kwargs": {"only_active": {"type": "bool"}},
+ },
+ "ghostwriter.shepherd.tasks.update_dns": {
+ "label": "Update DNS Records",
+ "args": [
+ {
+ "name": "domain",
+ "type": "int",
+ "nullable": True,
+ "min": 1,
+ "required": False,
+ }
+ ],
+ "kwargs": {"domain": {"type": "int", "nullable": True, "min": 1}},
+ },
+ "ghostwriter.modules.oplog_monitors.review_active_logs": {
+ "label": "Review Active Operation Logs",
+ "args": [
+ {
+ "name": "hours",
+ "type": "int",
+ "min": 1,
+ "max": 8760,
+ "required": False,
+ }
+ ],
+ "kwargs": {"hours": {"type": "int", "min": 1, "max": 8760}},
+ },
+ "ghostwriter.home.django_q_tasks.clear_expired_sessions": {
+ "label": "Clear Expired Sessions",
+ "args": [],
+ "kwargs": {},
+ },
+}
+
+# These tasks are queued by Ghostwriter itself. They are accepted by the queue
+# worker but are intentionally omitted from the schedule admin choices.
+GHOSTWRITER_DJANGO_Q_INTERNAL_TASKS = {
+ "ghostwriter.shepherd.tasks.namecheap_reset_dns": {"allow_any_arguments": True},
+ "ghostwriter.shepherd.tasks.test_aws_keys": {"allow_any_arguments": True},
+ "ghostwriter.shepherd.tasks.test_digital_ocean": {"allow_any_arguments": True},
+ "ghostwriter.shepherd.tasks.test_namecheap": {"allow_any_arguments": True},
+ "ghostwriter.shepherd.tasks.test_slack_webhook": {"allow_any_arguments": True},
+ "ghostwriter.shepherd.tasks.test_virustotal": {"allow_any_arguments": True},
+}
+
+# Hooks have their own execution path in Django Q and must be independently
+# constrained. The built-in Slack completion hook is safe to expose to schedules.
+GHOSTWRITER_DJANGO_Q_INTERNAL_HOOKS = {
+ "ghostwriter.modules.notifications_slack.send_slack_complete_msg": {
+ "label": "Send Slack Completion Message"
+ }
+}
+GHOSTWRITER_DJANGO_Q_SCHEDULE_HOOKS = {
+ "ghostwriter.modules.notifications_slack.send_slack_complete_msg": {
+ "label": "Send Slack Completion Message"
+ }
+}
+
+# Optional fixed commands configured by a server operator. When this mapping is
+# non-empty, the schedule admin exposes the restricted command runner. Each
+# command must use a fixed absolute argv and is always executed with shell=False.
+GHOSTWRITER_DJANGO_Q_COMMANDS = {}
+
# SETTINGS
# ------------------------------------------------------------------------------
# All settings are stored in singleton models in the CommandCenter app
diff --git a/ghostwriter/home/admin.py b/ghostwriter/home/admin.py
index bba29f88e..18c38d859 100644
--- a/ghostwriter/home/admin.py
+++ b/ghostwriter/home/admin.py
@@ -1,15 +1,27 @@
"""This contains customizations for displaying the Home application models in the admin panel."""
-# Standard Library Imports
+# Standard Libraries
import os
# Django Imports
+from django import forms
from django.conf import settings
-from django.contrib import admin
+from django.contrib import admin, messages
from django.urls import reverse
from django.utils.html import format_html
+# 3rd Party Libraries
+from django_q import admin as django_q_admin
+from django_q.models import Failure, Schedule, Success
+from django_q.tasks import async_task
+
# Ghostwriter Libraries
+from ghostwriter.home.django_q_policy import (
+ TaskPolicyError,
+ get_hook_policy,
+ get_schedule_policy,
+ validate_schedule,
+)
from ghostwriter.home.models import UserProfile
@@ -39,3 +51,121 @@ def avatar_download_link(self, obj):
)
return "File missing or not available for download"
avatar_download_link.short_description = "Download File"
+
+
+class RestrictedScheduleForm(forms.ModelForm):
+ """Expose only server-approved functions and hooks for scheduled tasks."""
+
+ func = forms.ChoiceField(label="Function")
+ hook = forms.ChoiceField(label="Hook", required=False)
+
+ class Meta:
+ model = Schedule
+ fields = "__all__"
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ task_policy = get_schedule_policy()
+ self.fields["func"].choices = [
+ (path, specification.get("label", path))
+ for path, specification in sorted(
+ task_policy.items(),
+ key=lambda item: item[1].get("label", item[0]),
+ )
+ ]
+ hook_policy = get_hook_policy(schedule_only=True)
+ self.fields["hook"].choices = [("", "---------")] + [
+ (path, specification.get("label", path))
+ for path, specification in sorted(
+ hook_policy.items(),
+ key=lambda item: item[1].get("label", item[0]),
+ )
+ ]
+
+ if self.instance and self.instance.pk:
+ if self.instance.func not in task_policy:
+ self.fields["func"].choices.append(
+ (self.instance.func, f"Disallowed: {self.instance.func}")
+ )
+ if self.instance.hook and self.instance.hook not in hook_policy:
+ self.fields["hook"].choices.append(
+ (self.instance.hook, f"Disallowed: {self.instance.hook}")
+ )
+
+ def clean(self):
+ cleaned_data = super().clean()
+ if self.errors:
+ return cleaned_data
+ self.instance.func = cleaned_data.get("func")
+ self.instance.hook = cleaned_data.get("hook") or None
+ self.instance.args = cleaned_data.get("args")
+ self.instance.kwargs = cleaned_data.get("kwargs")
+ self.instance.intended_date_kwarg = cleaned_data.get("intended_date_kwarg")
+ try:
+ validate_schedule(self.instance)
+ except TaskPolicyError as error:
+ raise forms.ValidationError(str(error)) from error
+ return cleaned_data
+
+
+@admin.action(description="Resubmit policy-approved tasks to queue")
+def resubmit_allowed_tasks(model_admin, request, queryset):
+ """Resubmit historical tasks only when current server policy permits them."""
+ submitted = 0
+ denied = 0
+ for task in queryset:
+ try:
+ async_task(
+ task.func,
+ *(task.args or ()),
+ hook=task.hook,
+ group=task.group,
+ cluster=task.cluster,
+ **(task.kwargs or {}),
+ )
+ except TaskPolicyError:
+ denied += 1
+ continue
+ submitted += 1
+ if isinstance(model_admin, RestrictedFailAdmin):
+ task.delete()
+ if submitted:
+ model_admin.message_user(
+ request,
+ f"Resubmitted {submitted} approved task(s).",
+ level=messages.SUCCESS,
+ )
+ if denied:
+ model_admin.message_user(
+ request,
+ f"Blocked {denied} task(s) that are not permitted by server policy.",
+ level=messages.ERROR,
+ )
+
+
+class RestrictedTaskAdmin(django_q_admin.TaskAdmin):
+ """Read-only successful task history with policy-aware resubmission."""
+
+ actions = [resubmit_allowed_tasks]
+
+
+class RestrictedFailAdmin(django_q_admin.FailAdmin):
+ """Read-only failed task history with policy-aware resubmission."""
+
+ actions = [resubmit_allowed_tasks]
+
+
+class RestrictedScheduleAdmin(django_q_admin.ScheduleAdmin):
+ """Django Q schedule admin constrained by server-side policy."""
+
+ form = RestrictedScheduleForm
+ readonly_fields = ("intended_date_kwarg",)
+
+
+for django_q_model in (Schedule, Success, Failure):
+ if admin.site.is_registered(django_q_model):
+ admin.site.unregister(django_q_model)
+
+admin.site.register(Schedule, RestrictedScheduleAdmin)
+admin.site.register(Success, RestrictedTaskAdmin)
+admin.site.register(Failure, RestrictedFailAdmin)
diff --git a/ghostwriter/home/apps.py b/ghostwriter/home/apps.py
index f6aaf2ab1..257f506f3 100644
--- a/ghostwriter/home/apps.py
+++ b/ghostwriter/home/apps.py
@@ -12,3 +12,8 @@ def ready(self):
import ghostwriter.home.signals # noqa F401 isort:skip
except ImportError:
pass
+
+ # Ghostwriter Libraries
+ from ghostwriter.home.django_q_integration import install_django_q_restrictions
+
+ install_django_q_restrictions()
diff --git a/ghostwriter/home/django_q_cluster.py b/ghostwriter/home/django_q_cluster.py
new file mode 100644
index 000000000..01603317c
--- /dev/null
+++ b/ghostwriter/home/django_q_cluster.py
@@ -0,0 +1,219 @@
+"""Restricted Django Q scheduler and worker implementations.
+
+These functions preserve the scheduler and worker contracts from the pinned
+django-q2 1.10.0 dependency while inserting policy checks before imports and
+execution. Review them whenever that dependency is upgraded.
+"""
+
+# Standard Libraries
+import pydoc
+import traceback
+
+# Django Imports
+from django.utils import timezone
+
+# 3rd Party Libraries
+import django_q.scheduler as q_scheduler
+import django_q.worker as q_worker
+
+# Ghostwriter Libraries
+from ghostwriter.home.django_q_policy import (
+ TaskPolicyError,
+ validate_schedule,
+ validate_task,
+)
+
+
+def restricted_scheduler(broker=None):
+ """Create tasks from due schedules after applying server policy."""
+ if not broker:
+ broker = q_scheduler.get_broker()
+ q_scheduler.close_old_django_connections()
+
+ default_cluster = (
+ q_scheduler.db.models.Q(cluster__isnull=True)
+ if q_scheduler.Conf.CLUSTER_NAME == q_scheduler.Conf.PREFIX
+ else q_scheduler.db.models.Q(pk__in=[])
+ )
+ try:
+ with q_scheduler.db.transaction.atomic(
+ using=q_scheduler.db.router.db_for_write(q_scheduler.Schedule)
+ ):
+ schedules = (
+ q_scheduler.Schedule.objects.select_for_update()
+ .exclude(repeats=0)
+ .filter(next_run__lt=timezone.now())
+ .filter(
+ default_cluster
+ | q_scheduler.db.models.Q(cluster=q_scheduler.Conf.CLUSTER_NAME)
+ )
+ )
+ for schedule in schedules:
+ try:
+ with q_scheduler.db.transaction.atomic(
+ using=q_scheduler.db.router.db_for_write(q_scheduler.Schedule)
+ ):
+ _enqueue_schedule(schedule, broker)
+ except TaskPolicyError as error:
+ q_scheduler.Schedule.objects.filter(pk=schedule.pk).update(
+ repeats=0
+ )
+ q_scheduler.logger.error(
+ "Paused Django Q schedule %s because server policy denied it: %s",
+ schedule.pk,
+ error,
+ )
+ except Exception: # pylint: disable=broad-exception-caught
+ # Isolate unexpected failures so later schedules are still checked.
+ q_scheduler.logger.exception(
+ "Could not create a task from Django Q schedule %s",
+ schedule.pk,
+ )
+ except Exception: # pylint: disable=broad-exception-caught
+ # The scheduler is a process boundary and must survive backend failures.
+ q_scheduler.logger.exception("Could not inspect Django Q schedules")
+
+
+def _enqueue_schedule(schedule, broker):
+ """Enqueue one locked schedule and calculate its following run."""
+ args, kwargs = validate_schedule(schedule)
+ q_options = {}
+ if schedule.hook:
+ q_options["hook"] = schedule.hook
+
+ if schedule.schedule_type != schedule.ONCE:
+ next_run = schedule.next_run
+ while True:
+ next_run = schedule.calculate_next_run(next_run)
+ if q_scheduler.Conf.CATCH_UP or next_run > q_scheduler.localtime():
+ break
+ schedule.next_run = next_run
+ if schedule.repeats < -1:
+ schedule.repeats = -1
+ if schedule.repeats > 0:
+ schedule.repeats -= 1
+
+ q_options["cluster"] = schedule.cluster
+ if (
+ q_options["cluster"] is None
+ or q_options["cluster"] == q_scheduler.Conf.CLUSTER_NAME
+ ):
+ q_options["broker"] = broker
+ q_options["group"] = schedule.name or schedule.id
+ kwargs["q_options"] = q_options
+ schedule.task = q_scheduler.async_task(schedule.func, *args, **kwargs)
+ if not schedule.task:
+ q_scheduler.logger.error(
+ "%s failed to create a task from schedule %s",
+ q_scheduler.current_process().name,
+ schedule.name or schedule.id,
+ )
+ else:
+ q_scheduler.logger.info(
+ "%s created task %s from schedule %s",
+ q_scheduler.current_process().name,
+ q_scheduler.humanize(schedule.task),
+ schedule.name or schedule.id,
+ )
+
+ if schedule.schedule_type == schedule.ONCE:
+ if schedule.repeats < 0:
+ schedule.delete()
+ return
+ schedule.repeats = 0
+ schedule.save()
+
+
+def restricted_worker(task_queue, result_queue, timer, timeout=q_worker.Conf.TIMEOUT):
+ """Execute queued tasks only after validating them against server policy."""
+ proc_name = q_worker.current_process().name
+ q_worker.logger.info(
+ "%s ready for work at %s", proc_name, q_worker.current_process().pid
+ )
+ q_worker.post_spawn.send(sender="django_q", proc_name=proc_name)
+ if q_worker.setproctitle:
+ q_worker.setproctitle.setproctitle(f"qcluster {proc_name} idle")
+ task_count = 0
+ if timeout is None:
+ timeout = -1
+
+ for task in iter(task_queue.get, "STOP"):
+ timer.value = -1
+ task_count += 1
+ func = task["func"]
+ func_name = q_worker.get_func_repr(func)
+ task_name = task["name"]
+ q_worker.logger.info("%s processing %s '%s'", proc_name, task_name, func_name)
+ if q_worker.setproctitle:
+ q_worker.setproctitle.setproctitle(
+ f"qcluster {proc_name} processing {task_name} '{func_name}'"
+ )
+ if not task.get("sync", False):
+ q_worker.close_old_django_connections()
+ timer_value = task.pop("timeout", timeout)
+ timeout_error = False
+ policy_denied = False
+
+ try:
+ validate_task(
+ func,
+ args=task.get("args", ()),
+ kwargs=task.get("kwargs", {}),
+ hook=task.get("hook"),
+ )
+ if not callable(func):
+ func = pydoc.locate(func)
+ if func is None:
+ raise ValueError(f"Function {task['func']} is not defined")
+ q_worker.pre_execute.send(sender="django_q", func=func, task=task)
+ timer.value = timer_value
+ if timer.value != -1:
+ timer.value += 3
+ with q_worker.TimeoutHandler(timer_value):
+ response = func(*task["args"], **task["kwargs"])
+ result = (response, True)
+ except TaskPolicyError as error:
+ policy_denied = True
+ task["ack_failure"] = True
+ result = (f"Task blocked by server policy: {error}", False)
+ q_worker.logger.error(
+ "Blocked Django Q task %s (%s): %s",
+ task_name,
+ func_name,
+ error,
+ )
+ except ( # pylint: disable=broad-exception-caught
+ Exception,
+ q_worker.TimeoutException,
+ ) as error:
+ # Task callables may raise any exception; workers record the failure.
+ if isinstance(error, q_worker.TimeoutException):
+ timeout_error = True
+ result = (f"{error} : {traceback.format_exc()}", False)
+ if q_worker.error_reporter:
+ q_worker.error_reporter.report()
+ if task.get("sync", False):
+ q_worker.post_execute_in_worker.send(
+ sender="django_q", func=func, task=task
+ )
+ raise
+
+ with timer.get_lock():
+ task["result"] = result[0]
+ task["success"] = result[1]
+ task["stopped"] = timezone.now()
+ if not policy_denied:
+ q_worker.post_execute_in_worker.send(
+ sender="django_q", func=func, task=task
+ )
+ result_queue.put(task)
+ if timeout_error:
+ timer.value = 0
+ break
+ timer.value = -1
+ if q_worker.setproctitle:
+ q_worker.setproctitle.setproctitle(f"qcluster {proc_name} idle")
+ if task_count == q_worker.Conf.RECYCLE or q_worker.rss_check():
+ timer.value = -2
+ break
+ q_worker.logger.info("%s stopped doing work", proc_name)
diff --git a/ghostwriter/home/django_q_integration.py b/ghostwriter/home/django_q_integration.py
new file mode 100644
index 000000000..92259c67e
--- /dev/null
+++ b/ghostwriter/home/django_q_integration.py
@@ -0,0 +1,106 @@
+"""Install Ghostwriter's Django Q security controls."""
+
+# Standard Libraries
+import logging
+import pydoc
+
+# Django Imports
+from django.core.checks import Error, Tags, register
+from django.core.exceptions import ImproperlyConfigured, ValidationError
+from django.db.models.signals import post_save, pre_save
+
+# 3rd Party Libraries
+import django_q.cluster
+from django_q.models import Schedule, Task
+from django_q.signals import call_hook, pre_enqueue
+
+# Ghostwriter Libraries
+from ghostwriter.home.django_q_cluster import restricted_scheduler, restricted_worker
+from ghostwriter.home.django_q_policy import (
+ TaskPolicyError,
+ callable_path,
+ get_hook_policy,
+ validate_policy_configuration,
+ validate_schedule,
+ validate_task,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def validate_schedule_on_save(sender, instance, raw=False, **kwargs):
+ """Reject Schedule writes that do not satisfy server policy."""
+ if raw:
+ return
+ try:
+ validate_schedule(instance)
+ except TaskPolicyError as error:
+ raise ValidationError(str(error)) from error
+
+
+def validate_task_before_enqueue(sender, task, **kwargs):
+ """Reject every disallowed task before Django Q signs or enqueues it."""
+ validate_task(
+ task.get("func"),
+ args=task.get("args", ()),
+ kwargs=task.get("kwargs", {}),
+ hook=task.get("hook"),
+ )
+
+
+def call_allowed_hook(sender, instance, **kwargs):
+ """Call a result hook only when its exact path is server-approved."""
+ if not instance.hook:
+ return
+ try:
+ hook_path = callable_path(instance.hook)
+ if hook_path not in get_hook_policy():
+ raise TaskPolicyError(
+ f"{hook_path} is not permitted by the server hook policy"
+ )
+ hook = pydoc.locate(hook_path)
+ if not callable(hook):
+ raise TaskPolicyError(f"Approved hook {hook_path} could not be loaded")
+ hook(instance)
+ except TaskPolicyError as error:
+ logger.error("Blocked Django Q result hook for task %s: %s", instance.pk, error)
+ except Exception: # pylint: disable=broad-exception-caught
+ # Hooks are best-effort and must not disrupt task-result persistence.
+ logger.exception(
+ "Approved Django Q result hook failed for task %s", instance.pk
+ )
+
+
+@register(Tags.security)
+def check_django_q_policy(app_configs, **kwargs):
+ """Report malformed server policy through Django's system checks."""
+ return [
+ Error(message, id="ghostwriter.EQ001")
+ for message in validate_policy_configuration()
+ ]
+
+
+def install_django_q_restrictions():
+ """Connect validation signals and replace Django Q execution functions."""
+ policy_errors = validate_policy_configuration()
+ if policy_errors:
+ raise ImproperlyConfigured(
+ f"Invalid Django Q server policy: {'; '.join(policy_errors)}"
+ )
+ pre_save.connect(
+ validate_schedule_on_save,
+ sender=Schedule,
+ dispatch_uid="ghostwriter.django_q.validate_schedule",
+ )
+ pre_enqueue.connect(
+ validate_task_before_enqueue,
+ dispatch_uid="ghostwriter.django_q.validate_enqueue",
+ )
+ post_save.disconnect(call_hook, sender=Task)
+ post_save.connect(
+ call_allowed_hook,
+ sender=Task,
+ dispatch_uid="ghostwriter.django_q.call_allowed_hook",
+ )
+ django_q.cluster.worker = restricted_worker
+ django_q.cluster.scheduler = restricted_scheduler
diff --git a/ghostwriter/home/django_q_policy.py b/ghostwriter/home/django_q_policy.py
new file mode 100644
index 000000000..9ba17c388
--- /dev/null
+++ b/ghostwriter/home/django_q_policy.py
@@ -0,0 +1,477 @@
+"""Server-side policy enforcement for Django Q tasks and schedules."""
+
+# Standard Libraries
+import ast
+import inspect
+import re
+from collections.abc import Mapping, Sequence
+
+# Django Imports
+from django.conf import settings
+
+COMMAND_RUNNER = "ghostwriter.home.django_q_tasks.run_configured_command"
+MAX_SCHEDULE_ARGUMENT_LENGTH = 10000
+RESERVED_KWARGS = {"q_options"}
+SUPPORTED_TYPES = {"bool", "float", "int", "str"}
+DOTTED_PATH_PATTERN = re.compile(r"^[A-Za-z_]\w*(?:\.[A-Za-z_]\w*)+$")
+
+
+class TaskPolicyError(Exception):
+ """Raised when a task, hook, command, or argument violates policy."""
+
+
+def callable_path(value):
+ """Return a stable dotted path without importing an untrusted string."""
+ if isinstance(value, str):
+ path = value
+ elif (
+ inspect.isfunction(value) or inspect.ismethod(value) or inspect.isbuiltin(value)
+ ):
+ module = getattr(value, "__module__", None)
+ name = getattr(value, "__qualname__", None) or getattr(value, "__name__", None)
+ path = f"{module}.{name}" if module and name else ""
+ else:
+ path = ""
+ if not path or not DOTTED_PATH_PATTERN.fullmatch(path) or "" in path:
+ raise TaskPolicyError("Task callable must be an exact dotted path")
+ return path
+
+
+def parse_schedule_arguments(args_value, kwargs_value):
+ """Parse Django Q's literal args and kwargs formats without evaluating code."""
+ args = ()
+ kwargs = {}
+
+ for label, value in (
+ ("Arguments", args_value),
+ ("Keyword arguments", kwargs_value),
+ ):
+ if isinstance(value, str) and len(value) > MAX_SCHEDULE_ARGUMENT_LENGTH:
+ raise TaskPolicyError(
+ f"{label} exceed the maximum length of {MAX_SCHEDULE_ARGUMENT_LENGTH} characters"
+ )
+
+ if args_value:
+ if isinstance(args_value, str):
+ try:
+ args = ast.literal_eval(args_value)
+ except (RecursionError, SyntaxError, ValueError, TypeError) as error:
+ raise TaskPolicyError(
+ "Arguments must contain Python literals"
+ ) from error
+ else:
+ args = args_value
+ if not isinstance(args, tuple):
+ args = (args,)
+
+ if kwargs_value:
+ if isinstance(kwargs_value, Mapping):
+ kwargs = dict(kwargs_value)
+ elif isinstance(kwargs_value, str):
+ try:
+ parsed = ast.literal_eval(kwargs_value)
+ if not isinstance(parsed, Mapping):
+ raise ValueError
+ kwargs = dict(parsed)
+ except (
+ RecursionError,
+ SyntaxError,
+ ValueError,
+ TypeError,
+ ) as literal_error:
+ try:
+ call = ast.parse(f"f({kwargs_value})", mode="eval").body
+ keywords = call.keywords
+ if call.args or any(keyword.arg is None for keyword in keywords):
+ raise ValueError from literal_error
+ names = [keyword.arg for keyword in keywords]
+ if len(names) != len(set(names)):
+ raise ValueError from literal_error
+ kwargs = {
+ keyword.arg: ast.literal_eval(keyword.value)
+ for keyword in keywords
+ }
+ except (RecursionError, SyntaxError, ValueError, TypeError) as error:
+ raise TaskPolicyError(
+ "Keyword arguments must contain Python literals"
+ ) from error
+ else:
+ raise TaskPolicyError("Keyword arguments must be a mapping")
+
+ return args, kwargs
+
+
+def _get_mapping_setting(setting_name):
+ """Return a copy of a mapping-based policy setting."""
+ configured = getattr(settings, setting_name, {})
+ if not isinstance(configured, Mapping):
+ raise TaskPolicyError(f"{setting_name} must be a mapping")
+ return dict(configured)
+
+
+def get_command_policy():
+ """Return a copy of the configured fixed-command mapping."""
+ return _get_mapping_setting("GHOSTWRITER_DJANGO_Q_COMMANDS")
+
+
+def get_schedule_policy():
+ """Return configured schedule tasks, adding the command runner when needed."""
+ policy = _get_mapping_setting("GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS")
+ commands = get_command_policy()
+ if commands:
+ if not all(isinstance(name, str) and name for name in commands):
+ raise TaskPolicyError("Command names must be non-empty strings")
+ command_names = sorted(commands)
+ policy[COMMAND_RUNNER] = {
+ "label": "Run Approved System Command",
+ "args": [
+ {
+ "name": "command_name",
+ "type": "str",
+ "required": False,
+ "choices": command_names,
+ }
+ ],
+ "kwargs": {
+ "command_name": {
+ "type": "str",
+ "choices": command_names,
+ }
+ },
+ "required_parameters": ["command_name"],
+ }
+ return policy
+
+
+def get_queue_policy():
+ """Return the union of internal and administrator-schedulable tasks."""
+ policy = _get_mapping_setting("GHOSTWRITER_DJANGO_Q_INTERNAL_TASKS")
+ policy.update(get_schedule_policy())
+ return policy
+
+
+def get_hook_policy(schedule_only=False):
+ """Return hooks permitted for schedules or for the complete queue."""
+ policy = _get_mapping_setting("GHOSTWRITER_DJANGO_Q_SCHEDULE_HOOKS")
+ if not schedule_only:
+ internal_hooks = _get_mapping_setting("GHOSTWRITER_DJANGO_Q_INTERNAL_HOOKS")
+ policy = internal_hooks | policy
+ return policy
+
+
+def _matches_policy_type(value, expected):
+ """Return whether a policy value exactly matches its declared type."""
+ if expected == "float":
+ return type(value) in (float, int)
+ expected_type = {"bool": bool, "int": int, "str": str}.get(expected)
+ return expected_type is not None and type(value) is expected_type
+
+
+def _validate_policy_bounds(name, specification, expected):
+ """Validate min/max constraints before comparing them to task values."""
+ for bound in ("min", "max"):
+ if bound in specification and not _matches_policy_type(
+ specification[bound], expected
+ ):
+ raise TaskPolicyError(f"Policy {bound} for {name} must be a {expected}")
+ if (
+ "min" in specification
+ and "max" in specification
+ and specification["min"] > specification["max"]
+ ):
+ raise TaskPolicyError(
+ f"Policy for {name} has a minimum greater than its maximum"
+ )
+
+
+def _validate_value(name, value, specification):
+ if not isinstance(specification, Mapping):
+ raise TaskPolicyError(f"Policy for {name} must be a mapping")
+ expected = specification.get("type")
+ if expected not in SUPPORTED_TYPES:
+ raise TaskPolicyError(f"Policy for {name} has an unsupported type")
+ _validate_policy_bounds(name, specification, expected)
+ if value is None:
+ if specification.get("nullable", False):
+ return
+ raise TaskPolicyError(f"{name} may not be null")
+ if not _matches_policy_type(value, expected):
+ raise TaskPolicyError(f"{name} must be a {expected}")
+
+ if "choices" in specification and value not in specification["choices"]:
+ raise TaskPolicyError(f"{name} is not an approved value")
+ if "min" in specification and value < specification["min"]:
+ raise TaskPolicyError(f"{name} is below the allowed minimum")
+ if "max" in specification and value > specification["max"]:
+ raise TaskPolicyError(f"{name} exceeds the allowed maximum")
+
+
+def validate_arguments(task_path, args, kwargs, task_policy):
+ """Validate arguments for one exact task policy entry."""
+ task_specification = task_policy[task_path]
+ if not isinstance(task_specification, Mapping):
+ raise TaskPolicyError(f"Policy for {task_path} must be a mapping")
+ if RESERVED_KWARGS.intersection(kwargs):
+ raise TaskPolicyError(f"{task_path} may not override Django Q options")
+ if task_specification.get("allow_any_arguments", False):
+ return
+
+ positional_specs = task_specification.get("args", [])
+ keyword_specs = task_specification.get("kwargs", {})
+ if not isinstance(positional_specs, Sequence) or isinstance(positional_specs, str):
+ raise TaskPolicyError(
+ f"Positional argument policy for {task_path} must be a list"
+ )
+ if not isinstance(keyword_specs, Mapping):
+ raise TaskPolicyError(
+ f"Keyword argument policy for {task_path} must be a mapping"
+ )
+
+ required_args = sum(1 for item in positional_specs if item.get("required", True))
+ if len(args) < required_args or len(args) > len(positional_specs):
+ raise TaskPolicyError(
+ f"{task_path} received an unapproved number of positional arguments"
+ )
+ for index, value in enumerate(args):
+ _validate_value(f"argument {index + 1}", value, positional_specs[index])
+
+ positional_names = {
+ specification.get("name")
+ for specification in positional_specs[: len(args)]
+ if specification.get("name")
+ }
+ duplicated = positional_names.intersection(kwargs)
+ if duplicated:
+ raise TaskPolicyError(
+ f"{task_path} received arguments more than once: {', '.join(sorted(duplicated))}"
+ )
+
+ unexpected = set(kwargs) - set(keyword_specs)
+ if unexpected:
+ raise TaskPolicyError(
+ f"{task_path} received unapproved keyword arguments: {', '.join(sorted(unexpected))}"
+ )
+ for name, value_specification in keyword_specs.items():
+ if value_specification.get("required", False) and name not in kwargs:
+ raise TaskPolicyError(f"{task_path} requires keyword argument {name}")
+ if name in kwargs:
+ _validate_value(name, kwargs[name], value_specification)
+
+ provided_parameters = positional_names | set(kwargs)
+ missing_parameters = (
+ set(task_specification.get("required_parameters", [])) - provided_parameters
+ )
+ if missing_parameters:
+ raise TaskPolicyError(
+ f"{task_path} requires parameters: {', '.join(sorted(missing_parameters))}"
+ )
+
+
+def validate_task(func, args=(), kwargs=None, hook=None, schedule_only=False):
+ """Validate a task package without importing its callable or hook."""
+ task_path = callable_path(func)
+ policy = get_schedule_policy() if schedule_only else get_queue_policy()
+ if task_path not in policy:
+ raise TaskPolicyError(f"{task_path} is not permitted by the server task policy")
+ validate_arguments(task_path, tuple(args or ()), dict(kwargs or {}), policy)
+ if hook:
+ hook_path = callable_path(hook)
+ hook_policy = get_hook_policy(schedule_only=schedule_only)
+ if hook_path not in hook_policy:
+ raise TaskPolicyError(
+ f"{hook_path} is not permitted by the server hook policy"
+ )
+ return task_path
+
+
+def validate_schedule(schedule):
+ """Parse and validate a Django Q Schedule instance."""
+ if schedule.intended_date_kwarg:
+ raise TaskPolicyError("Scheduled tasks may not inject an intended-date keyword")
+ args, kwargs = parse_schedule_arguments(schedule.args, schedule.kwargs)
+ validate_task(
+ schedule.func,
+ args=args,
+ kwargs=kwargs,
+ hook=schedule.hook,
+ schedule_only=True,
+ )
+ return args, kwargs
+
+
+def _validate_command(name, command):
+ """Validate one fixed command definition."""
+ if not isinstance(name, str) or not name:
+ raise TaskPolicyError("Command names must be non-empty strings")
+ if not isinstance(command, Mapping):
+ raise TaskPolicyError(f"Command {name} must be a mapping")
+ argv = command.get("argv")
+ if (
+ not isinstance(argv, (list, tuple))
+ or not argv
+ or not all(isinstance(item, str) and item for item in argv)
+ ):
+ raise TaskPolicyError(f"Command {name} must define a non-empty string argv")
+ if not argv[0].startswith("/"):
+ raise TaskPolicyError(f"Command {name} must use an absolute executable path")
+ if "shell" in command:
+ raise TaskPolicyError(f"Command {name} may not configure a shell")
+ if "cwd" in command and (
+ not isinstance(command["cwd"], str) or not command["cwd"].startswith("/")
+ ):
+ raise TaskPolicyError(f"Command {name} working directory must be absolute")
+ if "timeout" in command and (
+ type(command["timeout"]) not in (float, int) or command["timeout"] <= 0
+ ):
+ raise TaskPolicyError(f"Command {name} timeout must be positive")
+ if "env" in command:
+ environment = command["env"]
+ if not isinstance(environment, Mapping) or not all(
+ isinstance(key, str) and isinstance(value, str)
+ for key, value in environment.items()
+ ):
+ raise TaskPolicyError(
+ f"Command {name} environment must contain only strings"
+ )
+
+
+def validate_command_policy():
+ """Validate fixed command definitions without executing them."""
+ for name, command in get_command_policy().items():
+ _validate_command(name, command)
+
+
+def _validate_policy_schema(path, specification):
+ """Validate the declarative structure of one task policy entry."""
+ if not isinstance(specification, Mapping):
+ raise TaskPolicyError(f"Policy for {path} must be a mapping")
+ if specification.get("allow_any_arguments", False):
+ return
+ positional = specification.get("args", [])
+ keywords = specification.get("kwargs", {})
+ if not isinstance(positional, Sequence) or isinstance(positional, str):
+ raise TaskPolicyError(f"Positional argument policy for {path} must be a list")
+ if not isinstance(keywords, Mapping):
+ raise TaskPolicyError(f"Keyword argument policy for {path} must be a mapping")
+ optional_seen = False
+ positional_names = []
+ for name, value_specification in [
+ *((f"argument {index + 1}", item) for index, item in enumerate(positional)),
+ *keywords.items(),
+ ]:
+ if not isinstance(value_specification, Mapping):
+ raise TaskPolicyError(f"Policy for {path} {name} must be a mapping")
+ expected = value_specification.get("type")
+ if expected not in SUPPORTED_TYPES:
+ raise TaskPolicyError(f"Policy for {path} {name} has an unsupported type")
+ _validate_policy_bounds(
+ f"{path} {name}",
+ value_specification,
+ expected,
+ )
+ if name.startswith("argument "):
+ parameter_name = value_specification.get("name")
+ if parameter_name is not None and (
+ not isinstance(parameter_name, str)
+ or not parameter_name
+ or parameter_name in positional_names
+ ):
+ raise TaskPolicyError(
+ f"Policy for {path} has an invalid or duplicate positional argument name"
+ )
+ if parameter_name:
+ positional_names.append(parameter_name)
+ required = value_specification.get("required", True)
+ if not required:
+ optional_seen = True
+ elif optional_seen:
+ raise TaskPolicyError(
+ f"Policy for {path} has a required argument after an optional one"
+ )
+ if "choices" in value_specification and (
+ not isinstance(value_specification["choices"], Sequence)
+ or isinstance(value_specification["choices"], str)
+ ):
+ raise TaskPolicyError(f"Policy choices for {path} {name} must be a list")
+
+ required_parameters = specification.get("required_parameters", [])
+ if (
+ not isinstance(required_parameters, Sequence)
+ or isinstance(required_parameters, str)
+ or not all(isinstance(name, str) for name in required_parameters)
+ or not set(required_parameters).issubset(set(positional_names) | set(keywords))
+ ):
+ raise TaskPolicyError(f"Policy for {path} has invalid required parameters")
+
+
+def _validate_task_policy_entry(path, specification):
+ """Validate one configured task path and its argument schema."""
+ callable_path(path)
+ _validate_policy_schema(path, specification)
+
+
+def _validate_hook_policy_entry(path, specification):
+ """Validate one configured hook path."""
+ callable_path(path)
+ if not isinstance(specification, Mapping):
+ raise TaskPolicyError(f"Hook policy for {path} must be a mapping")
+
+
+def _collect_policy_error(errors, validator, *args):
+ """Run one policy validation and append its error, if any."""
+ try:
+ return validator(*args)
+ except TaskPolicyError as error:
+ errors.append(str(error))
+ return None
+
+
+def validate_policy_configuration():
+ """Validate every configured policy entry and return all errors."""
+ errors = []
+ for setting_name in (
+ "GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS",
+ "GHOSTWRITER_DJANGO_Q_INTERNAL_TASKS",
+ ):
+ policy = _collect_policy_error(
+ errors,
+ _get_mapping_setting,
+ setting_name,
+ )
+ if policy is not None:
+ for path, specification in policy.items():
+ _collect_policy_error(
+ errors,
+ _validate_task_policy_entry,
+ path,
+ specification,
+ )
+
+ for setting_name in (
+ "GHOSTWRITER_DJANGO_Q_SCHEDULE_HOOKS",
+ "GHOSTWRITER_DJANGO_Q_INTERNAL_HOOKS",
+ ):
+ policy = _collect_policy_error(
+ errors,
+ _get_mapping_setting,
+ setting_name,
+ )
+ if policy is not None:
+ for path, specification in policy.items():
+ _collect_policy_error(
+ errors,
+ _validate_hook_policy_entry,
+ path,
+ specification,
+ )
+
+ commands = _collect_policy_error(
+ errors,
+ _get_mapping_setting,
+ "GHOSTWRITER_DJANGO_Q_COMMANDS",
+ )
+ if commands is not None:
+ for name, command in commands.items():
+ _collect_policy_error(errors, _validate_command, name, command)
+
+ return errors
diff --git a/ghostwriter/home/django_q_tasks.py b/ghostwriter/home/django_q_tasks.py
new file mode 100644
index 000000000..dea4f6b5b
--- /dev/null
+++ b/ghostwriter/home/django_q_tasks.py
@@ -0,0 +1,39 @@
+"""Narrow Django Q task wrappers approved by server-side policy."""
+
+# Standard Libraries
+import subprocess
+
+# Django Imports
+from django.core.management import call_command
+
+# Ghostwriter Libraries
+from ghostwriter.home.django_q_policy import (
+ TaskPolicyError,
+ get_command_policy,
+ validate_command_policy,
+)
+
+
+def clear_expired_sessions():
+ """Run only Ghostwriter's expired-session cleanup management command."""
+ call_command("clear_expired_sessions")
+
+
+def run_configured_command(command_name):
+ """Run one fixed server-approved command without a shell."""
+ validate_command_policy()
+ commands = get_command_policy()
+ if command_name not in commands:
+ raise TaskPolicyError("The requested command is not approved")
+ command = commands[command_name]
+ completed = subprocess.run(
+ list(command["argv"]),
+ cwd=command.get("cwd"),
+ env=dict(command.get("env", {})),
+ shell=False,
+ check=True,
+ timeout=command.get("timeout"),
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+ return {"returncode": completed.returncode}
diff --git a/ghostwriter/home/management/commands/audit_django_q_policy.py b/ghostwriter/home/management/commands/audit_django_q_policy.py
new file mode 100644
index 000000000..b01250b0e
--- /dev/null
+++ b/ghostwriter/home/management/commands/audit_django_q_policy.py
@@ -0,0 +1,56 @@
+"""Audit Django Q schedules against Ghostwriter's server-side policy."""
+
+# Django Imports
+from django.core.management.base import BaseCommand, CommandError
+
+# 3rd Party Libraries
+from django_q.models import Schedule
+
+# Ghostwriter Libraries
+from ghostwriter.home.django_q_policy import TaskPolicyError, validate_schedule
+
+
+class Command(BaseCommand):
+ help = "Audit Django Q schedules against the server task allowlist"
+
+ def add_arguments(self, parser):
+ parser.add_argument(
+ "--check",
+ action="store_true",
+ help="Exit with an error when a disallowed schedule is found",
+ )
+ parser.add_argument(
+ "--pause-disallowed",
+ action="store_true",
+ help="Set repeats=0 on every disallowed schedule",
+ )
+
+ def handle(self, *args, **options):
+ disallowed = []
+ for schedule in Schedule.objects.order_by("pk"):
+ try:
+ validate_schedule(schedule)
+ except TaskPolicyError as error:
+ disallowed.append((schedule, error))
+ self.stdout.write(
+ self.style.WARNING(
+ f"DISALLOWED schedule={schedule.pk} func={schedule.func}: {error}"
+ )
+ )
+ if options["pause_disallowed"] and schedule.repeats != 0:
+ Schedule.objects.filter(pk=schedule.pk).update(repeats=0)
+
+ if not disallowed:
+ self.stdout.write(
+ self.style.SUCCESS("All Django Q schedules satisfy server policy.")
+ )
+ return
+
+ if options["pause_disallowed"]:
+ self.stdout.write(
+ self.style.SUCCESS(f"Paused {len(disallowed)} disallowed schedule(s).")
+ )
+ if options["check"]:
+ raise CommandError(
+ f"Found {len(disallowed)} disallowed Django Q schedule(s)."
+ )
diff --git a/ghostwriter/home/tests/test_django_q_policy.py b/ghostwriter/home/tests/test_django_q_policy.py
new file mode 100644
index 000000000..470d3be7a
--- /dev/null
+++ b/ghostwriter/home/tests/test_django_q_policy.py
@@ -0,0 +1,920 @@
+"""Tests for Ghostwriter's Django Q task policy."""
+
+# Standard Libraries
+import inspect
+import pydoc
+import queue
+from contextlib import nullcontext
+from io import StringIO
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+# Django Imports
+from django.contrib import admin
+from django.core.exceptions import ImproperlyConfigured, ValidationError
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.db.models.signals import post_save
+from django.test import SimpleTestCase, TestCase, override_settings
+from django.utils import timezone
+
+# 3rd Party Libraries
+from django_q.conf import Conf
+from django_q.exceptions import TimeoutException
+from django_q.models import Failure, Schedule, Success, Task
+from django_q.signals import call_hook
+from django_q.tasks import async_task
+
+# Ghostwriter Libraries
+from ghostwriter.home import django_q_tasks
+from ghostwriter.home.admin import RestrictedScheduleAdmin, RestrictedScheduleForm
+from ghostwriter.home.django_q_cluster import restricted_scheduler, restricted_worker
+from ghostwriter.home.django_q_integration import (
+ call_allowed_hook,
+ install_django_q_restrictions,
+ validate_schedule_on_save,
+)
+from ghostwriter.home.django_q_policy import (
+ COMMAND_RUNNER,
+ TaskPolicyError,
+ callable_path,
+ get_schedule_policy,
+ parse_schedule_arguments,
+ validate_policy_configuration,
+ validate_schedule,
+ validate_task,
+)
+
+
+def allowed_test_task(value):
+ """Return a value for restricted-worker tests."""
+ return value
+
+
+def failing_test_task():
+ """Raise an ordinary task exception for restricted-worker tests."""
+ raise RuntimeError("expected task failure")
+
+
+def allowed_test_hook(task):
+ """Mark a task when an approved test hook is called."""
+ task.hook_called = True
+
+
+def failing_test_hook(task):
+ """Raise an ordinary hook exception for hook-isolation tests."""
+ raise RuntimeError(f"expected hook failure for {task.pk}")
+
+
+TEST_TASK_PATH = "ghostwriter.home.tests.test_django_q_policy.allowed_test_task"
+FAILING_TEST_TASK_PATH = "ghostwriter.home.tests.test_django_q_policy.failing_test_task"
+TEST_HOOK_PATH = "ghostwriter.home.tests.test_django_q_policy.allowed_test_hook"
+FAILING_TEST_HOOK_PATH = "ghostwriter.home.tests.test_django_q_policy.failing_test_hook"
+TEST_TASK_POLICY = {
+ TEST_TASK_PATH: {
+ "label": "Allowed Test Task",
+ "args": [{"type": "int"}],
+ "kwargs": {},
+ }
+}
+
+
+class FakeTimer:
+ """Provide the small multiprocessing.Value interface used by workers."""
+
+ def __init__(self):
+ self.value = -1
+
+ def get_lock(self):
+ return nullcontext()
+
+
+class TaskPolicyTests(SimpleTestCase):
+ def test_callable_path_accepts_function(self):
+ self.assertEqual(callable_path(allowed_test_task), TEST_TASK_PATH)
+
+ def test_callable_path_rejects_non_callable_and_malformed_values(self):
+ for value in (object(), "not-a-dotted-path"):
+ with self.subTest(value=value):
+ with self.assertRaisesRegex(TaskPolicyError, "exact dotted path"):
+ callable_path(value)
+
+ def test_builtin_task_is_allowed_with_approved_arguments(self):
+ path = validate_task(
+ "ghostwriter.shepherd.tasks.release_domains",
+ kwargs={"no_action": True},
+ schedule_only=True,
+ )
+
+ self.assertEqual(path, "ghostwriter.shepherd.tasks.release_domains")
+
+ def test_builtin_policy_exposes_every_callable_parameter(self):
+ for path, specification in get_schedule_policy().items():
+ function = pydoc.locate(path)
+ parameters = [
+ parameter.name
+ for parameter in inspect.signature(function).parameters.values()
+ if parameter.kind
+ in (parameter.POSITIONAL_ONLY, parameter.POSITIONAL_OR_KEYWORD)
+ ]
+ positional_names = [item.get("name") for item in specification["args"]]
+
+ self.assertEqual(positional_names, parameters, path)
+ self.assertEqual(set(specification["kwargs"]), set(parameters), path)
+
+ def test_all_builtin_parameters_are_allowed_positionally(self):
+ sample_values = {"bool": True, "int": 1, "float": 1.0, "str": "value"}
+ for path, specification in get_schedule_policy().items():
+ args = tuple(
+ sample_values[item["type"]] for item in specification.get("args", [])
+ )
+
+ validate_task(path, args=args, schedule_only=True)
+
+ def test_parameter_cannot_be_supplied_positionally_and_by_name(self):
+ with self.assertRaisesRegex(TaskPolicyError, "more than once"):
+ validate_task(
+ "ghostwriter.shepherd.tasks.release_domains",
+ args=(True,),
+ kwargs={"no_action": False},
+ schedule_only=True,
+ )
+
+ def test_arbitrary_callable_is_denied(self):
+ with self.assertRaises(TaskPolicyError):
+ validate_task("os.system", args=("id",), schedule_only=True)
+
+ @patch("django_q.tasks.get_broker")
+ def test_arbitrary_callable_is_denied_before_enqueue(self, get_broker_mock):
+ get_broker_mock.return_value = Mock(list_key="test")
+
+ with self.assertRaises(TaskPolicyError):
+ async_task("os.system", "id")
+
+ get_broker_mock.return_value.enqueue.assert_not_called()
+
+ def test_allowed_callable_rejects_unapproved_arguments(self):
+ with self.assertRaisesRegex(TaskPolicyError, "unapproved keyword"):
+ validate_task(
+ "ghostwriter.shepherd.tasks.release_domains",
+ kwargs={"command": "id"},
+ schedule_only=True,
+ )
+
+ def test_builtin_argument_constraints_are_enforced(self):
+ path = "ghostwriter.modules.oplog_monitors.review_active_logs"
+ for value, message in (
+ (True, "must be a int"),
+ (0, "below the allowed minimum"),
+ (8761, "exceeds the allowed maximum"),
+ ):
+ with self.subTest(value=value):
+ with self.assertRaisesRegex(TaskPolicyError, message):
+ validate_task(path, args=(value,), schedule_only=True)
+
+ validate_task(
+ "ghostwriter.shepherd.tasks.check_domains",
+ args=(None,),
+ schedule_only=True,
+ )
+
+ def test_django_q_options_are_always_denied(self):
+ with override_settings(
+ GHOSTWRITER_DJANGO_Q_INTERNAL_TASKS={
+ TEST_TASK_PATH: {"allow_any_arguments": True}
+ }
+ ):
+ with self.assertRaisesRegex(TaskPolicyError, "may not override"):
+ validate_task(TEST_TASK_PATH, kwargs={"q_options": {"timeout": 0}})
+
+ def test_schedule_arguments_use_literal_parsing(self):
+ args, kwargs = parse_schedule_arguments("1, 'two'", "enabled=True, count=3")
+
+ self.assertEqual(args, (1, "two"))
+ self.assertEqual(kwargs, {"enabled": True, "count": 3})
+
+ def test_schedule_arguments_accept_single_values_and_mapping_syntax(self):
+ args, kwargs = parse_schedule_arguments("1", "{'enabled': True}")
+
+ self.assertEqual(args, (1,))
+ self.assertEqual(kwargs, {"enabled": True})
+
+ def test_schedule_arguments_accept_python_values(self):
+ args, kwargs = parse_schedule_arguments((1, 2), {"enabled": True})
+
+ self.assertEqual(args, (1, 2))
+ self.assertEqual(kwargs, {"enabled": True})
+
+ def test_schedule_arguments_reject_code(self):
+ with self.assertRaises(TaskPolicyError):
+ parse_schedule_arguments("__import__('os').system('id')", "")
+
+ def test_schedule_arguments_reject_kwargs_expansion_and_duplicates(self):
+ for kwargs in (
+ "1",
+ "**{'enabled': True}",
+ "enabled=True, enabled=False",
+ ):
+ with self.subTest(kwargs=kwargs):
+ with self.assertRaisesRegex(
+ TaskPolicyError,
+ "Keyword arguments must contain Python literals",
+ ):
+ parse_schedule_arguments("", kwargs)
+
+ def test_schedule_arguments_reject_non_mapping_kwargs(self):
+ with self.assertRaisesRegex(TaskPolicyError, "must be a mapping"):
+ parse_schedule_arguments("", object())
+
+ def test_schedule_arguments_reject_excessive_input(self):
+ with self.assertRaisesRegex(TaskPolicyError, "maximum length"):
+ parse_schedule_arguments("'" + ("a" * 10000) + "'", "")
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_COMMANDS={
+ "backup": {
+ "argv": ["/usr/local/bin/ghostwriter-backup"],
+ "timeout": 30,
+ }
+ }
+ )
+ def test_fixed_commands_add_restricted_runner_to_schedule_policy(self):
+ policy = get_schedule_policy()
+
+ self.assertIn(COMMAND_RUNNER, policy)
+ validate_task(
+ COMMAND_RUNNER,
+ kwargs={"command_name": "backup"},
+ schedule_only=True,
+ )
+ validate_task(COMMAND_RUNNER, args=("backup",), schedule_only=True)
+ with self.assertRaisesRegex(TaskPolicyError, "requires parameters"):
+ validate_task(COMMAND_RUNNER, schedule_only=True)
+ with self.assertRaises(TaskPolicyError):
+ validate_task(
+ COMMAND_RUNNER,
+ kwargs={"command_name": "shell"},
+ schedule_only=True,
+ )
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_COMMANDS={
+ "backup": {
+ "argv": ["/usr/local/bin/ghostwriter-backup", "--quiet"],
+ "timeout": 30,
+ }
+ }
+ )
+ @patch("ghostwriter.home.django_q_tasks.subprocess.run")
+ def test_command_runner_uses_fixed_argv_without_shell(self, run_mock):
+ run_mock.return_value = SimpleNamespace(returncode=0)
+
+ result = django_q_tasks.run_configured_command("backup")
+
+ self.assertEqual(result, {"returncode": 0})
+ run_mock.assert_called_once_with(
+ ["/usr/local/bin/ghostwriter-backup", "--quiet"],
+ cwd=None,
+ env={},
+ shell=False,
+ check=True,
+ timeout=30,
+ stdout=-3,
+ stderr=-3,
+ )
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_COMMANDS={"bad": {"argv": ["sh", "-c", "id"]}}
+ )
+ def test_relative_command_policy_fails_system_check(self):
+ self.assertTrue(validate_policy_configuration())
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS={
+ "not a dotted path": {},
+ TEST_TASK_PATH: {
+ "args": [{"type": "int", "min": "1"}],
+ "kwargs": {},
+ },
+ }
+ )
+ def test_policy_configuration_reports_multiple_errors(self):
+ errors = validate_policy_configuration()
+
+ self.assertGreaterEqual(len(errors), 2)
+ self.assertTrue(
+ any("exact dotted path" in error for error in errors),
+ errors,
+ )
+ self.assertTrue(
+ any("Policy min" in error for error in errors),
+ errors,
+ )
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS={
+ TEST_TASK_PATH: {
+ "args": [{"type": "int", "min": "1"}],
+ "kwargs": {},
+ }
+ }
+ )
+ def test_invalid_policy_bound_fails_closed_at_runtime(self):
+ with self.assertRaisesRegex(TaskPolicyError, "Policy min"):
+ validate_task(TEST_TASK_PATH, args=(1,), schedule_only=True)
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS={
+ TEST_TASK_PATH: {
+ "args": [{"type": "int", "min": 5, "max": 1}],
+ "kwargs": {},
+ }
+ }
+ )
+ def test_policy_configuration_rejects_inverted_bounds(self):
+ self.assertTrue(
+ any(
+ "minimum greater than its maximum" in error
+ for error in validate_policy_configuration()
+ )
+ )
+
+ def test_command_policy_reports_each_invalid_definition(self):
+ invalid_commands = (
+ {1: {"argv": ["/bin/true"]}},
+ {"bad": "not-a-mapping"},
+ {"bad": {"argv": []}},
+ {"bad": {"argv": ["/bin/true"], "shell": False}},
+ {"bad": {"argv": ["/bin/true"], "cwd": "relative"}},
+ {"bad": {"argv": ["/bin/true"], "timeout": 0}},
+ {"bad": {"argv": ["/bin/true"], "env": {"KEY": 1}}},
+ )
+ for commands in invalid_commands:
+ with self.subTest(commands=commands):
+ with override_settings(GHOSTWRITER_DJANGO_Q_COMMANDS=commands):
+ self.assertTrue(validate_policy_configuration())
+
+ def test_policy_configuration_validates_each_schema_shape(self):
+ invalid_specifications = (
+ "not-a-mapping",
+ {"args": "not-a-list", "kwargs": {}},
+ {"args": [], "kwargs": "not-a-mapping"},
+ {"args": ["not-a-mapping"], "kwargs": {}},
+ {"args": [{"type": "unsupported"}], "kwargs": {}},
+ {
+ "args": [{"name": "", "type": "int"}],
+ "kwargs": {},
+ },
+ {
+ "args": [
+ {"name": "first", "type": "int", "required": False},
+ {"name": "second", "type": "int"},
+ ],
+ "kwargs": {},
+ },
+ {
+ "args": [{"type": "int", "choices": "not-a-list"}],
+ "kwargs": {},
+ },
+ {
+ "args": [],
+ "kwargs": {},
+ "required_parameters": ["missing"],
+ },
+ )
+ for specification in invalid_specifications:
+ with self.subTest(specification=specification):
+ with override_settings(
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS={TEST_TASK_PATH: specification}
+ ):
+ self.assertTrue(validate_policy_configuration())
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS=[],
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_HOOKS=[],
+ GHOSTWRITER_DJANGO_Q_COMMANDS=[],
+ )
+ def test_policy_configuration_aggregates_invalid_mapping_settings(self):
+ errors = validate_policy_configuration()
+
+ self.assertEqual(len(errors), 3)
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_HOOKS={TEST_HOOK_PATH: "not-a-mapping"}
+ )
+ def test_policy_configuration_rejects_invalid_hook_schema(self):
+ self.assertTrue(validate_policy_configuration())
+
+ @override_settings(GHOSTWRITER_DJANGO_Q_COMMANDS={1: {"argv": ["/bin/true"]}})
+ def test_schedule_policy_rejects_invalid_command_name(self):
+ with self.assertRaisesRegex(TaskPolicyError, "Command names"):
+ get_schedule_policy()
+
+ @override_settings(GHOSTWRITER_DJANGO_Q_COMMANDS={})
+ @patch("ghostwriter.home.django_q_tasks.call_command")
+ def test_clear_expired_sessions_uses_narrow_management_command(
+ self, call_command_mock
+ ):
+ django_q_tasks.clear_expired_sessions()
+
+ call_command_mock.assert_called_once_with("clear_expired_sessions")
+
+ def test_command_runner_rejects_unknown_command(self):
+ with self.assertRaisesRegex(TaskPolicyError, "not approved"):
+ django_q_tasks.run_configured_command("missing")
+
+ def test_schedule_rejects_intended_date_keyword(self):
+ schedule = Schedule(
+ func="ghostwriter.home.django_q_tasks.clear_expired_sessions",
+ intended_date_kwarg="run_at",
+ )
+
+ with self.assertRaisesRegex(TaskPolicyError, "intended-date"):
+ validate_schedule(schedule)
+
+ def test_task_rejects_unapproved_hook(self):
+ with self.assertRaisesRegex(TaskPolicyError, "hook policy"):
+ validate_task(
+ "ghostwriter.home.django_q_tasks.clear_expired_sessions",
+ hook="os.system",
+ schedule_only=True,
+ )
+
+
+class RestrictedAdminTests(SimpleTestCase):
+ def test_django_q_models_use_restricted_admins(self):
+ self.assertIsInstance(admin.site._registry[Schedule], RestrictedScheduleAdmin)
+ self.assertNotEqual(
+ admin.site._registry[Success].actions[0].__name__,
+ "resubmit_task",
+ )
+ self.assertNotEqual(
+ admin.site._registry[Failure].actions[0].__name__,
+ "resubmit_task",
+ )
+
+ def test_schedule_form_rejects_forged_callable(self):
+ form = RestrictedScheduleForm(
+ data={
+ "name": "Malicious",
+ "func": "os.system",
+ "hook": "",
+ "args": "'id'",
+ "kwargs": "",
+ "schedule_type": Schedule.DAILY,
+ "minutes": "",
+ "repeats": -1,
+ "next_run": timezone.now(),
+ "cron": "",
+ "cluster": "",
+ "intended_date_kwarg": "",
+ }
+ )
+
+ self.assertFalse(form.is_valid())
+ self.assertIn("func", form.errors)
+
+ def test_schedule_form_exposes_builtin_slack_hook(self):
+ form = RestrictedScheduleForm()
+
+ self.assertIn(
+ (
+ "ghostwriter.modules.notifications_slack.send_slack_complete_msg",
+ "Send Slack Completion Message",
+ ),
+ list(form.fields["hook"].choices),
+ )
+
+ def test_schedule_admin_allows_cluster_routing_only(self):
+ model_admin = admin.site._registry[Schedule]
+
+ self.assertNotIn("cluster", model_admin.readonly_fields)
+ self.assertIn("cluster", RestrictedScheduleForm().fields)
+ self.assertIn("intended_date_kwarg", model_admin.readonly_fields)
+
+ def test_schedule_form_accepts_approved_callable(self):
+ form = RestrictedScheduleForm(
+ data={
+ "name": "Release Domains",
+ "func": "ghostwriter.shepherd.tasks.release_domains",
+ "hook": "",
+ "args": "",
+ "kwargs": "no_action=True",
+ "schedule_type": Schedule.DAILY,
+ "minutes": "",
+ "repeats": -1,
+ "next_run": timezone.now(),
+ "cron": "",
+ "cluster": "",
+ "intended_date_kwarg": "",
+ }
+ )
+
+ self.assertTrue(form.is_valid(), form.errors.as_json())
+
+
+class SchedulePolicyIntegrationTests(TestCase):
+ def test_model_save_signal_rejects_disallowed_schedule(self):
+ with self.assertRaises(ValidationError) as raised:
+ Schedule.objects.create(
+ name="Malicious",
+ func="os.system",
+ args="'id'",
+ schedule_type=Schedule.DAILY,
+ )
+
+ self.assertFalse(hasattr(raised.exception, "error_dict"))
+ self.assertIn("os.system is not permitted", raised.exception.messages[0])
+
+ def test_audit_command_reports_clean_policy(self):
+ stdout = StringIO()
+
+ call_command("audit_django_q_policy", "--check", stdout=stdout)
+
+ self.assertIn("All Django Q schedules satisfy", stdout.getvalue())
+
+ def test_audit_command_reports_and_pauses_bulk_inserted_schedule(self):
+ schedule = Schedule(
+ name="Malicious",
+ func="os.system",
+ args="'id'",
+ schedule_type=Schedule.DAILY,
+ repeats=-1,
+ )
+ Schedule.objects.bulk_create([schedule])
+ stdout = StringIO()
+
+ with self.assertRaises(CommandError):
+ call_command("audit_django_q_policy", "--check", stdout=stdout)
+ self.assertIn("DISALLOWED", stdout.getvalue())
+
+ call_command("audit_django_q_policy", "--pause-disallowed", stdout=StringIO())
+ schedule.refresh_from_db()
+ self.assertEqual(schedule.repeats, 0)
+
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.close_old_django_connections")
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.get_broker")
+ def test_scheduler_uses_configured_broker(
+ self, get_broker_mock, _close_connections
+ ):
+ get_broker_mock.return_value = Mock()
+
+ restricted_scheduler()
+
+ get_broker_mock.assert_called_once_with()
+
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.close_old_django_connections")
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.async_task")
+ def test_scheduler_pauses_denied_row_and_continues_allowed_rows(
+ self, async_task_mock, _close_connections
+ ):
+ async_task_mock.return_value = "0123456789abcdef0123456789abcdef"
+ invalid = Schedule(
+ name="Malicious",
+ func="os.system",
+ args="'id'",
+ schedule_type=Schedule.DAILY,
+ repeats=-1,
+ next_run=timezone.now(),
+ cluster=Conf.CLUSTER_NAME,
+ )
+ Schedule.objects.bulk_create([invalid])
+ valid = Schedule.objects.create(
+ name="Cleanup",
+ func="ghostwriter.home.django_q_tasks.clear_expired_sessions",
+ schedule_type=Schedule.ONCE,
+ repeats=1,
+ next_run=timezone.now(),
+ cluster=Conf.CLUSTER_NAME,
+ )
+
+ restricted_scheduler(broker=Mock())
+
+ invalid.refresh_from_db()
+ valid.refresh_from_db()
+ self.assertEqual(invalid.repeats, 0)
+ self.assertEqual(valid.repeats, 0)
+ async_task_mock.assert_called_once()
+
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.logger")
+ @patch("ghostwriter.home.django_q_cluster._enqueue_schedule")
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.close_old_django_connections")
+ def test_scheduler_isolates_unexpected_schedule_failure(
+ self, _close_connections, enqueue_mock, logger_mock
+ ):
+ for name in ("First", "Second"):
+ Schedule.objects.create(
+ name=name,
+ func="ghostwriter.home.django_q_tasks.clear_expired_sessions",
+ schedule_type=Schedule.DAILY,
+ repeats=-1,
+ next_run=timezone.now(),
+ cluster=Conf.CLUSTER_NAME,
+ )
+ enqueue_mock.side_effect = [RuntimeError("expected failure"), None]
+
+ restricted_scheduler(broker=Mock())
+
+ self.assertEqual(enqueue_mock.call_count, 2)
+ logger_mock.exception.assert_called_once()
+
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.logger")
+ @patch(
+ "ghostwriter.home.django_q_cluster.q_scheduler.db.transaction.atomic",
+ side_effect=RuntimeError("database unavailable"),
+ )
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.close_old_django_connections")
+ def test_scheduler_logs_backend_failure(
+ self, _close_connections, _atomic, logger_mock
+ ):
+ restricted_scheduler(broker=Mock())
+
+ logger_mock.exception.assert_called_once_with(
+ "Could not inspect Django Q schedules"
+ )
+
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.close_old_django_connections")
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.async_task")
+ def test_scheduler_advances_recurring_schedule(
+ self, async_task_mock, _close_connections
+ ):
+ async_task_mock.return_value = "0123456789abcdef0123456789abcdef"
+ original_next_run = timezone.now()
+ schedule = Schedule.objects.create(
+ name="Recurring cleanup",
+ func="ghostwriter.home.django_q_tasks.clear_expired_sessions",
+ schedule_type=Schedule.DAILY,
+ repeats=2,
+ next_run=original_next_run,
+ cluster=Conf.CLUSTER_NAME,
+ )
+
+ restricted_scheduler(broker=Mock())
+
+ schedule.refresh_from_db()
+ self.assertEqual(schedule.repeats, 1)
+ self.assertGreater(schedule.next_run, original_next_run)
+
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.close_old_django_connections")
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.async_task")
+ def test_scheduler_preserves_approved_hook_and_normalizes_repeats(
+ self, async_task_mock, _close_connections
+ ):
+ async_task_mock.return_value = "0123456789abcdef0123456789abcdef"
+ schedule = Schedule.objects.create(
+ name="Hooked cleanup",
+ func="ghostwriter.home.django_q_tasks.clear_expired_sessions",
+ hook="ghostwriter.modules.notifications_slack.send_slack_complete_msg",
+ schedule_type=Schedule.DAILY,
+ repeats=-2,
+ next_run=timezone.now(),
+ cluster=Conf.CLUSTER_NAME,
+ )
+
+ restricted_scheduler(broker=Mock())
+
+ schedule.refresh_from_db()
+ self.assertEqual(schedule.repeats, -1)
+ q_options = async_task_mock.call_args.kwargs["q_options"]
+ self.assertEqual(q_options["hook"], schedule.hook)
+
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.logger")
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.close_old_django_connections")
+ @patch(
+ "ghostwriter.home.django_q_cluster.q_scheduler.async_task",
+ return_value=None,
+ )
+ def test_scheduler_records_enqueue_failure(
+ self, _async_task, _close_connections, logger_mock
+ ):
+ schedule = Schedule.objects.create(
+ name="Failed enqueue",
+ func="ghostwriter.home.django_q_tasks.clear_expired_sessions",
+ schedule_type=Schedule.ONCE,
+ repeats=1,
+ next_run=timezone.now(),
+ cluster=Conf.CLUSTER_NAME,
+ )
+
+ restricted_scheduler(broker=Mock())
+
+ logger_mock.error.assert_called_once()
+ schedule.refresh_from_db()
+ self.assertEqual(schedule.repeats, 0)
+
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.close_old_django_connections")
+ @patch("ghostwriter.home.django_q_cluster.q_scheduler.async_task")
+ def test_scheduler_deletes_completed_one_time_schedule(
+ self, async_task_mock, _close_connections
+ ):
+ async_task_mock.return_value = "0123456789abcdef0123456789abcdef"
+ schedule = Schedule.objects.create(
+ name="One time cleanup",
+ func="ghostwriter.home.django_q_tasks.clear_expired_sessions",
+ schedule_type=Schedule.ONCE,
+ repeats=-1,
+ next_run=timezone.now(),
+ cluster=Conf.CLUSTER_NAME,
+ )
+
+ restricted_scheduler(broker=Mock())
+
+ self.assertFalse(Schedule.objects.filter(pk=schedule.pk).exists())
+
+
+class WorkerPolicyTests(SimpleTestCase):
+ def test_raw_schedule_save_skips_policy_validation(self):
+ validate_schedule_on_save(
+ Schedule,
+ Schedule(func="os.system"),
+ raw=True,
+ )
+
+ def test_result_without_hook_needs_no_policy_check(self):
+ call_allowed_hook(None, SimpleNamespace(pk="task-id", hook=None))
+
+ @override_settings(GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS=[])
+ def test_install_restrictions_rejects_invalid_policy(self):
+ with self.assertRaisesRegex(ImproperlyConfigured, "server policy"):
+ install_django_q_restrictions()
+
+ def test_unrestricted_django_q_hook_receiver_is_disconnected(self):
+ synchronous, asynchronous = post_save._live_receivers(Task)
+
+ self.assertNotIn(call_hook, [*synchronous, *asynchronous])
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_INTERNAL_TASKS=TEST_TASK_POLICY,
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS={},
+ )
+ @patch("ghostwriter.home.django_q_cluster.q_worker.rss_check", return_value=False)
+ @patch("ghostwriter.home.django_q_cluster.q_worker.setproctitle", None)
+ def test_worker_executes_approved_task(self, _rss_check):
+ task_queue = queue.Queue()
+ result_queue = queue.Queue()
+ task_queue.put(
+ {
+ "id": "allowed-id",
+ "name": "allowed",
+ "func": TEST_TASK_PATH,
+ "args": (7,),
+ "kwargs": {},
+ "started": timezone.now(),
+ }
+ )
+ task_queue.put("STOP")
+
+ restricted_worker(task_queue, result_queue, FakeTimer())
+
+ result = result_queue.get_nowait()
+ self.assertTrue(result["success"])
+ self.assertEqual(result["result"], 7)
+
+ @patch("ghostwriter.home.django_q_cluster.pydoc.locate")
+ @patch("ghostwriter.home.django_q_cluster.q_worker.rss_check", return_value=False)
+ @patch("ghostwriter.home.django_q_cluster.q_worker.setproctitle", None)
+ def test_worker_denies_task_before_import_and_acknowledges_failure(
+ self, _rss_check, locate_mock
+ ):
+ task_queue = queue.Queue()
+ result_queue = queue.Queue()
+ task_queue.put(
+ {
+ "id": "blocked-id",
+ "name": "blocked",
+ "func": "os.system",
+ "args": ("id",),
+ "kwargs": {},
+ "started": timezone.now(),
+ }
+ )
+ task_queue.put("STOP")
+
+ restricted_worker(task_queue, result_queue, FakeTimer())
+
+ result = result_queue.get_nowait()
+ self.assertFalse(result["success"])
+ self.assertTrue(result["ack_failure"])
+ self.assertIn("blocked by server policy", result["result"])
+ locate_mock.assert_not_called()
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_INTERNAL_TASKS={
+ FAILING_TEST_TASK_PATH: {"args": [], "kwargs": {}}
+ },
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS={},
+ )
+ @patch("ghostwriter.home.django_q_cluster.q_worker.rss_check", return_value=False)
+ @patch("ghostwriter.home.django_q_cluster.q_worker.setproctitle", None)
+ def test_worker_records_approved_task_exception(self, _rss_check):
+ task_queue = queue.Queue()
+ result_queue = queue.Queue()
+ task_queue.put(
+ {
+ "id": "failure-id",
+ "name": "failure",
+ "func": FAILING_TEST_TASK_PATH,
+ "args": (),
+ "kwargs": {},
+ "started": timezone.now(),
+ }
+ )
+ task_queue.put("STOP")
+
+ restricted_worker(task_queue, result_queue, FakeTimer())
+
+ result = result_queue.get_nowait()
+ self.assertFalse(result["success"])
+ self.assertIn("expected task failure", result["result"])
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_INTERNAL_TASKS=TEST_TASK_POLICY,
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS={},
+ )
+ @patch(
+ "ghostwriter.home.django_q_cluster.q_worker.TimeoutHandler",
+ side_effect=TimeoutException,
+ )
+ @patch("ghostwriter.home.django_q_cluster.q_worker.setproctitle", None)
+ def test_worker_records_timeout_and_stops(self, _timeout_handler):
+ task_queue = queue.Queue()
+ result_queue = queue.Queue()
+ timer = FakeTimer()
+ task_queue.put(
+ {
+ "id": "timeout-id",
+ "name": "timeout",
+ "func": TEST_TASK_PATH,
+ "args": (7,),
+ "kwargs": {},
+ "started": timezone.now(),
+ }
+ )
+
+ restricted_worker(task_queue, result_queue, timer)
+
+ result = result_queue.get_nowait()
+ self.assertFalse(result["success"])
+ self.assertEqual(timer.value, 0)
+
+ @override_settings(
+ GHOSTWRITER_DJANGO_Q_INTERNAL_TASKS={
+ FAILING_TEST_TASK_PATH: {"args": [], "kwargs": {}}
+ },
+ GHOSTWRITER_DJANGO_Q_SCHEDULE_TASKS={},
+ )
+ @patch("ghostwriter.home.django_q_cluster.q_worker.setproctitle", None)
+ def test_synchronous_worker_reraises_task_exception(self):
+ task_queue = queue.Queue()
+ task_queue.put(
+ {
+ "id": "sync-failure-id",
+ "name": "sync failure",
+ "func": FAILING_TEST_TASK_PATH,
+ "args": (),
+ "kwargs": {},
+ "sync": True,
+ "started": timezone.now(),
+ }
+ )
+
+ with self.assertRaisesRegex(RuntimeError, "expected task failure"):
+ restricted_worker(task_queue, queue.Queue(), FakeTimer())
+
+ @patch("ghostwriter.home.django_q_integration.pydoc.locate")
+ def test_result_hook_is_denied_before_import(self, locate_mock):
+ task = SimpleNamespace(pk="task-id", hook="os.system")
+
+ call_allowed_hook(None, task)
+
+ locate_mock.assert_not_called()
+
+ @override_settings(GHOSTWRITER_DJANGO_Q_INTERNAL_HOOKS={TEST_HOOK_PATH: {}})
+ @patch("ghostwriter.home.django_q_integration.logger")
+ @patch("ghostwriter.home.django_q_integration.pydoc.locate", return_value=None)
+ def test_missing_approved_result_hook_is_blocked(self, _locate, logger_mock):
+ task = SimpleNamespace(pk="task-id", hook=TEST_HOOK_PATH)
+
+ call_allowed_hook(None, task)
+
+ logger_mock.error.assert_called_once()
+
+ @override_settings(GHOSTWRITER_DJANGO_Q_INTERNAL_HOOKS={TEST_HOOK_PATH: {}})
+ def test_approved_result_hook_runs(self):
+ task = SimpleNamespace(pk="task-id", hook=TEST_HOOK_PATH)
+
+ call_allowed_hook(None, task)
+
+ self.assertTrue(task.hook_called)
+
+ @override_settings(GHOSTWRITER_DJANGO_Q_INTERNAL_HOOKS={FAILING_TEST_HOOK_PATH: {}})
+ @patch("ghostwriter.home.django_q_integration.logger")
+ def test_approved_result_hook_failure_is_isolated(self, logger_mock):
+ task = SimpleNamespace(pk="task-id", hook=FAILING_TEST_HOOK_PATH)
+
+ call_allowed_hook(None, task)
+
+ logger_mock.exception.assert_called_once_with(
+ "Approved Django Q result hook failed for task %s",
+ "task-id",
+ )