From 260b72c98d01e4d4e30ab9a35f790bc6d1492278 Mon Sep 17 00:00:00 2001 From: Marco D'Aleo Date: Sat, 31 Oct 2026 15:22:46 +0000 Subject: [PATCH] Add managed cron lifecycle and environment support --- README.md | 5 + SECURITY.md | 2 + docs/architecture.md | 9 +- docs/cli.md | 25 +- docs/cron.md | 60 +++- docs/security-model.md | 16 +- pyproject.toml | 2 +- src/schedls/backends/base.py | 3 + src/schedls/backends/cron.py | 260 ++++++++++++-- src/schedls/backends/systemd.py | 3 + src/schedls/cli.py | 69 +++- src/schedls/models.py | 1 + src/schedls/operations/inspect.py | 3 + src/schedls/renderers/cron.py | 133 ++++++- tests/unit/test_cli.py | 95 ++++- tests/unit/test_cron_system.py | 569 +++++++++++++++++++++++++++++- tests/unit/test_crontab.py | 65 ++++ tests/unit/test_operations.py | 79 ++++- tests/unit/test_prompt.py | 93 ++++- tests/unit/test_renderers_cron.py | 58 +++ 20 files changed, 1457 insertions(+), 93 deletions(-) diff --git a/README.md b/README.md index 1c82d7a..daaea90 100644 --- a/README.md +++ b/README.md @@ -114,8 +114,13 @@ Create a cron job: ```console $ schedls new cleanup --cron --cron-expr '0 4 * * 0' -- /usr/local/bin/cleanup +$ schedls new backup --cron --daily 02:00 --env TZ=UTC -- /usr/local/bin/backup ``` +Cron `--env` values apply only to that managed job; they are scheduler +configuration rather than secret storage. Managed cron jobs can also be enabled +or disabled without deleting their saved schedule. + Create a system cron drop-in as root (runs as `root` unless `--run-as` is given): ```console diff --git a/SECURITY.md b/SECURITY.md index 51c72e1..5b5d4a6 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -38,6 +38,7 @@ Before every release: [ ] No pager/editor invocation under privilege [ ] Unit serialization tests pass [ ] Cron serialization tests pass +[ ] Managed cron state and environment parsing tests pass [ ] Symlink tests pass [ ] Atomic-write tests pass [ ] Trusted-directory ownership/mode tests pass @@ -49,4 +50,5 @@ Before every release: [ ] New mutating operations support --dry-run [ ] New mutating operations have clear confirmation behavior [ ] Documentation describes new filesystem/state changes +[ ] Environment values are documented as non-secret scheduler configuration ``` diff --git a/docs/architecture.md b/docs/architecture.md index 8c5627a..f3588c2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -98,7 +98,10 @@ time to detect concurrent edits. ## JSON schema `schedls --json` emits a single document on stdout; diagnostics go to stderr. -The schema version is `1`. +The schema version is `1`. Its field names are a compatibility contract: this +release adds no JSON keys or schema version. Managed cron jobs use the existing +`enabled` field (`true` for active, `false` for disabled); unmanaged or unknown +cron state remains `null`. ```json { @@ -128,3 +131,7 @@ field names; missing data is `null`, never invented. Jobs whose `backend` is `cron` additionally carry a `cron` object (`expression`, `shell`, `mailto`, `line`, `user`, `environment`), where `user` is the run-as account for system entries and `null` for the current user's crontab. + +Managed cron environments use that existing `cron.environment` array. No +environment values are added to unrelated cron entries or to a file-wide +document field. diff --git a/docs/cli.md b/docs/cli.md index 5f45ad3..b29d020 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -127,6 +127,9 @@ everything else keeps its current value. Passing `--` without `--command` is an error, so the command can never be changed by accident. - `--no-persistent` clears `Persistent=` on a systemd timer. +- `--env` supplied during an edit replaces the complete environment. Omitting it + preserves the current values; `--clear-env` removes every value instead. + `--env` and `--clear-env` cannot be combined. - Editing is supported for schedls-managed systemd timers and cron jobs. A cron edit changes only the job line in its managed block and preserves every other cron line. Editing a system cron drop-in requires root and preserves its @@ -150,9 +153,9 @@ $ schedls enable backup $ schedls disable backup ``` -Enable or disable a systemd timer. Cron has no portable enabled/disabled -concept, so these commands explain that limitation and exit without changing -anything. +Enable or disable a systemd timer or a schedls-managed cron job. For cron, +`schedls` comments or restores the managed job line without changing its +schedule or command; unmanaged cron jobs remain immutable. ### `logs NAME` @@ -241,7 +244,8 @@ and [cron.md](cron.md#schedule-shortcuts). | `--jitter DURATION` | systemd only | `--jitter 5min` | | `--accuracy DURATION` | systemd only | `--accuracy 1min` | | `--working-directory PATH` | systemd only | `--working-directory /srv/data` | -| `--env KEY=VALUE` | systemd only | `--env TZ=UTC` (repeatable) | +| `--env KEY=VALUE` | both | `--env TZ=UTC` (repeatable) | +| `--clear-env` | both, edit only| remove every environment value | | `--shell SCRIPT` | both | `--shell 'echo hi \| tee /tmp/log'` | Notes: @@ -253,8 +257,14 @@ Notes: `--cron-expr` with `--timer`, or `--persistent` with `--cron`) instead of being silently ignored. - Cron accepts a single schedule. -- Cron environment variables are not supported yet; passing `--env` with - `--cron` is rejected rather than silently ignored. +- `--env` may be repeated. On `new`, it sets the complete environment. On + `edit`, omitting it preserves the current environment, while supplying it + replaces every value. `--clear-env` explicitly selects an empty environment; + it is valid only on `edit` and cannot be combined with `--env`. +- For cron, values are execution-local shell assignments on the managed job + line, never file-wide crontab assignments. In system cron, they follow the + run-as account. See [cron command rendering](cron.md#command-rendering) for + `--shell` pipeline scope. - Cron jobs are created for the current user by default. With `--system` (root only), schedls creates a system cron drop-in; use `--run-as USER` to choose the account that runs it. @@ -291,6 +301,9 @@ $ schedls edit backup -i - `new` asks for the scope (user or system) unless `--user`/`--system` was given. Choosing system for a cron job also asks which user it should run as (`--run-as`, default `root`). +- The wizard can add environment values when creating a job and replace them + when editing one. Choosing replacement and entering no values clears the + environment. - A terminal is required. With piped or redirected input, `schedls` exits `3` and asks you to pass flags instead. - Interactive mode cannot be combined with `--json` (exit `2`). diff --git a/docs/cron.md b/docs/cron.md index 0638961..a6d87e3 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -58,6 +58,19 @@ block: If markers are malformed or a `begin` has no matching `end`, mutations refuse to proceed rather than guess. +Existing active blocks need no migration. A disabled managed job keeps the same +block and replaces its job line with a column-zero comment: + +```cron +# schedls:begin name=backup +# schedls:disabled 0 2 * * * /usr/local/bin/backup /srv/data +# schedls:end name=backup +``` + +The literal `# schedls:disabled ` prefix is the disabled representation; the +unprefixed line is enabled. This keeps the job discoverable and reversible +without giving cron a new file-wide setting. + ## Editing `schedls edit` supports jobs in valid schedls-managed blocks. It updates only @@ -69,6 +82,11 @@ When only the schedule changes, schedls retains the existing cron command without reverse-parsing or re-quoting shell syntax. Replacing a command uses the normal safe command renderer. +When an environment is replaced or cleared, schedls preserves the existing raw +command field and replaces only its managed assignment prefix. A malformed or +non-canonical managed prefix produces a warning and makes mutation refuse +rather than guess at shell syntax. + For a system drop-in, editing requires root and re-checks the expected `/etc/cron.d/schedls-` path, ownership, trusted directory, and file snapshot before atomically replacing the file. The existing `run-as` user is @@ -98,6 +116,32 @@ Use `--run-as USER` to run a system job as another account (default `root`). The interactive wizard (`schedls new -i`) asks for the scope, and choosing the system scope for a cron job then prompts for the run-as user. +## Per-job environment + +`--env KEY=VALUE` creates execution-local POSIX shell assignments on a managed +job line. They are not file-wide crontab environment declarations, so they do +not affect unrelated jobs. A user job is rendered like this: + +```cron +0 2 * * * TZ=UTC /usr/local/bin/backup +``` + +For a system job, the run-as account stays before the assignments: + +```cron +0 2 * * * backup TZ=UTC /usr/local/bin/backup +``` + +`--env` may be repeated. On creation it supplies the complete environment. On +edit, omitting it preserves the current values; supplying one or more entries +replaces the complete set; and `--clear-env` explicitly removes every value. +`--env` and `--clear-env` are mutually exclusive. The interactive wizard offers +the same choices, including an empty replacement to clear the environment. + +Environment values are shell-quoted and percent-escaped with the command field. +They are scheduler configuration, not secret storage: current-user crontabs and +system cron drop-ins may be readable by other users. + ## Command rendering Cron executes command text through a shell. `schedls` keeps an argv internally @@ -113,6 +157,11 @@ representation exists. Raw shell syntax is available explicitly with `--shell`. `schedls` never infers shell mode from characters such as `|`, `>`, `&&` or `$()`. +For cron `--shell`, a managed environment is an ordinary leading POSIX +assignment. In a pipeline such as `TZ=UTC producer | consumer`, it applies to +the first command only; schedls does not add a wrapper to widen the assignment's +scope. Export a value in the script when later pipeline commands need it. + ## Schedule shortcuts | Flag | Example | Compiles to | @@ -147,8 +196,15 @@ CronCapabilities( ## Enable / disable -Cron has no universal native enabled/disabled concept. `schedls enable` and -`schedls disable` support systemd only and explain this limitation for cron. +`schedls enable` and `schedls disable` support schedls-managed cron jobs using +the disabled-line representation above. They are idempotent and only add or +remove that prefix, preserving the saved schedule and command text exactly. +Unmanaged cron jobs are never changed. + +Older schedls versions do not understand the disabled-line representation. They +will leave its job inactive because cron treats it as a comment, but cannot edit +that block correctly. Re-enable the job with a version that supports this +format before downgrading if you need to manage it with an older release. ## Logs diff --git a/docs/security-model.md b/docs/security-model.md index 2700063..2792ed5 100644 --- a/docs/security-model.md +++ b/docs/security-model.md @@ -77,8 +77,9 @@ of JSON encoding. Helper commands run with a controlled environment (`LC_ALL=C`, `SYSTEMD_COLORS=0`, `SYSTEMD_PAGER=cat`, `PAGER=cat`) and never with a shell. Pagers and editors are never launched while `schedls` is running. Environment -values passed with `--env` are written to unit files with systemd quoting and -are world-readable like any other unit; do not put secrets in them. +values passed with `--env` are written to unit files with systemd quoting or to +managed cron command prefixes with shell quoting. They are scheduler +configuration that may be readable by other users; do not put secrets in them. ### TOCTOU races @@ -89,11 +90,12 @@ practical. ### Corrupt or malicious existing configuration -Configuration is treated as opaque data. Malformed `schedls` cron markers cause -mutation to fail closed rather than guess. Unknown crontab lines are preserved -byte-for-byte. A systemd unit whose name starts with `schedls-` is only treated -as managed when its file carries the `# Managed by schedls` and `# Name:` -markers, so a same-named foreign unit is not modified. Likewise a +Configuration is treated as opaque data. Malformed `schedls` cron markers or +managed environment prefixes cause mutation to fail closed rather than guess. +Unknown crontab lines are preserved byte-for-byte. A systemd unit whose name +starts with `schedls-` is only treated as managed when its file carries the +`# Managed by schedls` and `# Name:` markers, so a same-named foreign unit is +not modified. Likewise a `/etc/cron.d/schedls-` drop-in is only managed when its begin/end markers match the file name; otherwise it is listed as unmanaged and left alone. diff --git a/pyproject.toml b/pyproject.toml index 1f6409f..228a720 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "schedls" -version = "0.4.1" +version = "0.5.0" description = "Inspect and manage Linux scheduled jobs" authors = [ { name = "Marco D'Aleo", email = "marco@marcodaleo.com" } diff --git a/src/schedls/backends/base.py b/src/schedls/backends/base.py index 344faa1..877854a 100644 --- a/src/schedls/backends/base.py +++ b/src/schedls/backends/base.py @@ -73,6 +73,9 @@ class SchedulerBackend: def discover(self, scopes: Sequence[Scope]) -> list[ScheduledJob]: raise NotImplementedError + def discovery_warnings(self) -> Sequence[str]: + return () + def find(self, name: str) -> ScheduledJob | None: for job in self.discover([Scope.USER, Scope.SYSTEM]): if job.name == name: diff --git a/src/schedls/backends/cron.py b/src/schedls/backends/cron.py index 8d00bc2..d6787e4 100644 --- a/src/schedls/backends/cron.py +++ b/src/schedls/backends/cron.py @@ -92,6 +92,9 @@ class CrontabEntry: name: str | None = None expression: str | None = None command: str | None = None + raw_command: str | None = None + environment: tuple[tuple[str, str], ...] = () + environment_malformed: bool = False user: str | None = None marker_name: str | None = None extra: dict[str, str] = field(default_factory=dict) @@ -181,7 +184,13 @@ class CrontabDocument: open_begin = None open_lines = [] elif open_begin is not None: - if entry.kind in {"job", "comment", "env", "blank"}: + if entry.kind == "disabled_malformed": + self.warnings.append(f"line {number}: malformed schedls disabled job") + open_lines.append(entry) + elif entry.kind in {"job", "disabled_job"} and entry.environment_malformed: + self.warnings.append(f"line {number}: malformed schedls environment prefix") + open_lines.append(entry) + elif entry.kind in {"job", "disabled_job", "comment", "env", "blank"}: open_lines.append(entry) if open_begin is not None: self.warnings.append( @@ -192,6 +201,12 @@ class CrontabDocument: stripped = raw.strip() if not stripped: return CrontabEntry(number, raw, "blank") + disabled_marker = renderer.DISABLED_PREFIX.rstrip() + if stripped == disabled_marker or stripped.startswith(renderer.DISABLED_PREFIX): + if not raw.startswith(renderer.DISABLED_PREFIX): + return CrontabEntry(number, raw, "disabled_malformed") + entry = self._job_entry(number, raw, raw[len(renderer.DISABLED_PREFIX) :].strip(), "disabled_job") + return entry or CrontabEntry(number, raw, "disabled_malformed") if stripped.startswith("#"): match = _MARKER_RE.match(stripped) if match: @@ -202,10 +217,24 @@ class CrontabDocument: return CrontabEntry(number, raw, "comment") if _ENV_RE.match(stripped): return CrontabEntry(number, raw, "env") - expression, command, user = _split_job(stripped, system=self.system) - if expression is None: - return CrontabEntry(number, raw, "comment") - return CrontabEntry(number, raw, "job", expression=expression, command=command, user=user) + return self._job_entry(number, raw, stripped, "job") or CrontabEntry(number, raw, "comment") + + def _job_entry(self, number: int, raw: str, text: str, kind: str) -> CrontabEntry | None: + expression, command, user = _split_job(text, system=self.system) + if expression is None or command is None: + return None + parsed = renderer.parse_environment_prefix(command) + return CrontabEntry( + number, + raw, + kind, + expression=expression, + command=parsed.command, + raw_command=command, + environment=parsed.environment, + environment_malformed=parsed.malformed, + user=user, + ) # -- queries -------------------------------------------------------------- @@ -258,7 +287,7 @@ class CrontabDocument: block = self._blocks.get(name) if block is None: raise SafetyRefusalError(f"no schedls-managed cron block named {name!r}.") - job_lines = [entry for entry in block.job_lines if entry.kind == "job"] + job_lines = [entry for entry in block.job_lines if entry.kind in {"job", "disabled_job"}] if len(job_lines) != 1: raise SafetyRefusalError( f"schedls-managed cron block {name!r} must contain exactly one job; refusing to edit." @@ -269,6 +298,20 @@ class CrontabDocument: job = self.managed_job_line(name) if has_unsafe_control_characters(line): raise SafetyRefusalError("cron job line must not contain NUL or newline") + if job.kind == "disabled_job": + line = renderer.render_disabled_line(line) + lines = list(self.lines) + lines[job.index - 1] = line + body = "\n".join(lines) + if self.had_trailing_newline and body: + return body + "\n" + return body + + def with_toggled_enabled(self, name: str, enabled: bool) -> str: + job = self.managed_job_line(name) + if (enabled and job.kind == "job") or (not enabled and job.kind == "disabled_job"): + return self.text + line = renderer.render_enabled_line(job.raw) if enabled else renderer.render_disabled_line(job.raw) lines = list(self.lines) lines[job.index - 1] = line body = "\n".join(lines) @@ -306,6 +349,8 @@ class CronBackend(SchedulerBackend): create=True, update=True, remove=True, + enable=True, + disable=True, next_run=False, validation=True, ) @@ -314,6 +359,7 @@ class CronBackend(SchedulerBackend): self.runner = runner self.paths = paths or CronPaths() self._capabilities: CronCapabilities | None = None + self._discovery_warnings: list[str] = [] def crontab_available(self) -> bool: return self.runner.has("crontab") @@ -381,6 +427,7 @@ class CronBackend(SchedulerBackend): # -- discovery ------------------------------------------------------------ def discover(self, scopes: Sequence[Scope]) -> list[ScheduledJob]: + self._discovery_warnings = [] jobs: list[ScheduledJob] = [] if Scope.USER in scopes: jobs.extend(self._discover_user_crontab()) @@ -390,21 +437,25 @@ class CronBackend(SchedulerBackend): jobs.extend(self._discover_periodic_directories()) return jobs + def discovery_warnings(self) -> Sequence[str]: + return tuple(self._discovery_warnings) + def _discover_user_crontab(self) -> list[ScheduledJob]: if not self.crontab_available(): return [] document = self.read() + self._record_document_warnings("current user's crontab", document) jobs: list[ScheduledJob] = [] managed_indices: set[int] = set() for name, block in document.managed_blocks().items(): managed_indices.update(range(block.begin_index, block.end_index + 1)) - job_line = next((line for line in block.job_lines if line.kind == "job"), None) + job_line = next((line for line in block.job_lines if line.kind in {"job", "disabled_job"}), None) job = ScheduledJob( name=name, backend=Backend.CRON, scope=Scope.USER, managed=True, - enabled=None, + enabled=job_line.kind == "job" if job_line else None, schedule=Schedule(ScheduleKind.CRON, (job_line.expression or "") if job_line else ""), command=Command(argv=(), raw=job_line.command if job_line else ""), source=JobSource("current user's crontab", line=block.begin_index), @@ -412,6 +463,7 @@ class CronBackend(SchedulerBackend): cron=CronDetails( expression=job_line.expression if job_line else None, line=block.begin_index, + environment=job_line.environment if job_line else (), ), ) jobs.append(job) @@ -426,7 +478,7 @@ class CronBackend(SchedulerBackend): managed=False, enabled=None, schedule=Schedule(ScheduleKind.CRON, entry.expression or ""), - command=Command(argv=(), raw=entry.command or ""), + command=Command(argv=(), raw=entry.raw_command or entry.command or ""), source=JobSource("current user's crontab", line=entry.index), cron=CronDetails( expression=entry.expression, @@ -452,6 +504,7 @@ class CronBackend(SchedulerBackend): if text is None: return [] document = CrontabDocument(text, system=True) + self._record_document_warnings(path, document) base = os.path.basename(path) blocks = document.managed_blocks() managed_name: str | None = None @@ -476,14 +529,14 @@ class CronBackend(SchedulerBackend): if managed_name is not None: block = blocks[managed_name] managed_range = set(range(block.begin_index, block.end_index + 1)) - job_line = next((line for line in block.job_lines if line.kind == "job"), None) + job_line = next((line for line in block.job_lines if line.kind in {"job", "disabled_job"}), None) jobs.append( ScheduledJob( name=managed_name, backend=Backend.CRON, scope=Scope.SYSTEM, managed=True, - enabled=None, + enabled=job_line.kind == "job" if job_line else None, schedule=Schedule(ScheduleKind.CRON, (job_line.expression or "") if job_line else ""), command=Command(argv=(), raw=job_line.command if job_line else ""), source=JobSource("system cron drop-in", path=path, line=block.begin_index), @@ -491,6 +544,7 @@ class CronBackend(SchedulerBackend): expression=job_line.expression if job_line else None, raw_line=job_line.raw if job_line else None, line=block.begin_index, + environment=job_line.environment if job_line else (), user=job_line.user if job_line else None, ), ) @@ -506,7 +560,7 @@ class CronBackend(SchedulerBackend): managed=False, enabled=None, schedule=Schedule(ScheduleKind.CRON, entry.expression or ""), - command=Command(argv=(), raw=entry.command or ""), + command=Command(argv=(), raw=entry.raw_command or entry.command or ""), source=JobSource("system cron", path=path, line=entry.index), cron=CronDetails( expression=entry.expression, @@ -517,6 +571,8 @@ class CronBackend(SchedulerBackend): warnings=tuple(warnings), ) ) + if warnings and not jobs: + self._discovery_warnings.extend(warnings) return jobs def _discover_periodic_directories(self) -> list[ScheduledJob]: @@ -576,7 +632,7 @@ class CronBackend(SchedulerBackend): managed=False, enabled=None, schedule=Schedule(ScheduleKind.CRON, entry.expression or ""), - command=Command(argv=(), raw=entry.command or ""), + command=Command(argv=(), raw=entry.raw_command or entry.command or ""), source=JobSource(f"user crontab ({name})", path=path, line=entry.index), cron=CronDetails( expression=entry.expression, @@ -602,6 +658,9 @@ class CronBackend(SchedulerBackend): except KeyError: return "" + def _record_document_warnings(self, source: str, document: CrontabDocument) -> None: + self._discovery_warnings.extend(f"{source}: {warning}" for warning in document.warnings) + @staticmethod def _list_dir(directory: str) -> list[str]: try: @@ -620,7 +679,7 @@ class CronBackend(SchedulerBackend): if not self.crontab_available(): raise OperationalError("crontab is not available") expression = renderer.validate_expression(spec.cron_expression or "") - line = renderer.render_line(expression, spec.command) + line = renderer.render_line(expression, spec.command, environment=spec.environment) document = self.read() new_text = document.with_block(spec.name, [line]) plan = Plan(backend=self.name, action="create") @@ -655,7 +714,7 @@ class CronBackend(SchedulerBackend): check_trusted_directory(directory, expected_uid=0) expression = renderer.validate_expression(spec.cron_expression or "") run_as = renderer.validate_run_as(spec.run_as or "root") - line = renderer.render_system_line(expression, spec.command, run_as) + line = renderer.render_system_line(expression, spec.command, run_as, environment=spec.environment) content = renderer.render_system_file(spec.name, [line]) path = os.path.join(directory, f"{_MANAGED_PREFIX}{spec.name}") if os.path.lexists(path): @@ -705,13 +764,18 @@ class CronBackend(SchedulerBackend): document = self.read() old_line = document.managed_job_line(current.name) expression = renderer.validate_expression(spec.cron_expression or "") + environment = spec.environment if spec.replace_environment else old_line.environment line = self._render_updated_line( expression, spec.command, + old_line.raw_command, old_line.command, preserve_command=spec.command == current.command, + existing_environment=old_line.environment, + environment=environment, ) new_text = document.with_updated_job(current.name, line) + expected_line = CrontabDocument(new_text).managed_job_line(current.name).raw plan = Plan(backend=self.name, action="update") plan.summary = [ ("Backend", "cron (current user)"), @@ -727,7 +791,7 @@ class CronBackend(SchedulerBackend): "name": current.name, "old_text": document.text, "new_text": new_text, - "expected_line": line, + "expected_line": expected_line, } return plan @@ -753,14 +817,19 @@ class CronBackend(SchedulerBackend): old_line = document.managed_job_line(current.name) run_as = renderer.validate_run_as(old_line.user or "") expression = renderer.validate_expression(spec.cron_expression or "") + environment = spec.environment if spec.replace_environment else old_line.environment line = self._render_updated_line( expression, spec.command, + old_line.raw_command, old_line.command, run_as=run_as, preserve_command=spec.command == current.command, + existing_environment=old_line.environment, + environment=environment, ) new_text = document.with_updated_job(current.name, line) + expected_line = CrontabDocument(new_text, system=True).managed_job_line(current.name).raw plan = Plan(backend=self.name, action="update") plan.summary = [ ("Backend", "cron (system)"), @@ -778,7 +847,7 @@ class CronBackend(SchedulerBackend): "path": path, "expected_uid": 0, "snapshots": {path: content}, - "expected_line": line, + "expected_line": expected_line, } return plan @@ -786,20 +855,137 @@ class CronBackend(SchedulerBackend): def _render_updated_line( expression: str, command: Command, + existing_command_field: str | None, existing_command: str | None, *, preserve_command: bool, + existing_environment: tuple[tuple[str, str], ...], + environment: tuple[tuple[str, str], ...], run_as: str | None = None, ) -> str: - if preserve_command: - if not existing_command or has_unsafe_control_characters(existing_command): + if preserve_command and environment == existing_environment: + if not existing_command_field or has_unsafe_control_characters(existing_command_field): raise SafetyRefusalError("existing cron command cannot be safely preserved.") if run_as is None: - return f"{expression} {existing_command}" - return f"{expression} {run_as} {existing_command}" + return f"{expression} {existing_command_field}" + return f"{expression} {run_as} {existing_command_field}" + if preserve_command: + return renderer.render_line_from_raw_command( + expression, + existing_command or "", + environment=environment, + user=run_as, + ) if run_as is None: - return renderer.render_line(expression, command) - return renderer.render_system_line(expression, command, run_as) + return renderer.render_line(expression, command, environment=environment) + return renderer.render_system_line(expression, command, run_as, environment=environment) + + def plan_set_enabled(self, job: ScheduledJob, enabled: bool) -> Plan: + if not job.managed: + raise SafetyRefusalError( + f"{job.name} was not created by schedls.", + hint="schedls will not modify unmanaged schedules by default.", + ) + if job.backend is not Backend.CRON: + raise SafetyRefusalError(f"{job.name} is not a cron job.") + if job.scope is Scope.USER: + return self._plan_set_enabled_user(job, enabled) + if job.scope is Scope.SYSTEM: + return self._plan_set_enabled_system(job, enabled) + raise SafetyRefusalError("unsupported cron scope.") + + def _noop_set_enabled_plan( + self, + job: ScheduledJob, + enabled: bool, + *, + mode: str, + old_text: str | None = None, + snapshots: dict[str, str | None] | None = None, + ) -> Plan: + state = "enabled" if enabled else "disabled" + plan = Plan(backend=self.name, action="enable" if enabled else "disable") + plan.summary = [("Backend", f"cron ({job.scope.value})"), ("State", f"already {state}")] + plan.payload = { + "mode": "noop", + "noop_mode": mode, + "action": plan.action, + "name": job.name, + "old_text": old_text, + "snapshots": snapshots or {}, + } + return plan + + def _plan_set_enabled_user(self, job: ScheduledJob, enabled: bool) -> Plan: + if not self.crontab_available(): + raise OperationalError("crontab is not available") + document = self.read() + if (document.managed_job_line(job.name).kind == "job") is enabled: + return self._noop_set_enabled_plan(job, enabled, mode="crontab", old_text=document.text) + new_text = document.with_toggled_enabled(job.name, enabled) + expected_line = CrontabDocument(new_text).managed_job_line(job.name).raw + action = "enable" if enabled else "disable" + plan = Plan(backend=self.name, action=action) + plan.summary = [ + ("Backend", "cron (current user)"), + ("State", "enabled" if enabled else "disabled"), + ] + plan.commands = self._install_commands(new_text) + plan.payload = { + "mode": "crontab", + "action": action, + "name": job.name, + "old_text": document.text, + "new_text": new_text, + "expected_line": expected_line, + } + return plan + + def _plan_set_enabled_system(self, job: ScheduledJob, enabled: bool) -> Plan: + if os.geteuid() != 0: + raise SafetyRefusalError( + "changing a system cron job requires appropriate privileges.", + hint=( + "Run the command under sudo yourself if that is your intention:\n" + " sudo schedls enable NAME\n" + " sudo schedls disable NAME" + ), + ) + renderer.validate_system_name(job.name) + directory = self.paths.cron_d + if not os.path.isdir(directory): + raise OperationalError(f"system cron directory does not exist: {directory}") + check_trusted_directory(directory, expected_uid=0) + path = job.source.path + expected_path = os.path.join(directory, f"{_MANAGED_PREFIX}{job.name}") + if path != expected_path: + raise SafetyRefusalError(f"refusing to edit unexpected system cron file: {path!r}") + content = _read_text(path) + if content is None: + raise OperationalError(f"could not read {path}") + document = CrontabDocument(content, system=True) + if (document.managed_job_line(job.name).kind == "job") is enabled: + return self._noop_set_enabled_plan(job, enabled, mode="file", snapshots={path: content}) + new_text = document.with_toggled_enabled(job.name, enabled) + expected_line = CrontabDocument(new_text, system=True).managed_job_line(job.name).raw + action = "enable" if enabled else "disable" + plan = Plan(backend=self.name, action=action) + plan.summary = [ + ("Backend", "cron (system)"), + ("State", "enabled" if enabled else "disabled"), + ("File", path), + ] + plan.files = [FileChange(path=path, content=new_text, mode=0o644, expected_uid=0)] + plan.payload = { + "mode": "file", + "action": action, + "name": job.name, + "path": path, + "expected_uid": 0, + "snapshots": {path: content}, + "expected_line": expected_line, + } + return plan def plan_remove(self, job: ScheduledJob) -> Plan: if not job.managed: @@ -864,10 +1050,24 @@ class CronBackend(SchedulerBackend): # -- applying ------------------------------------------------------------- def apply(self, plan: Plan) -> MutationResult: + if plan.payload.get("mode") == "noop": + return self._apply_noop(plan) if plan.payload.get("mode") == "file": return self._apply_file(plan) return self._apply_crontab(plan) + def _apply_noop(self, plan: Plan) -> MutationResult: + if plan.payload.get("noop_mode") == "crontab": + if self.read().text != plan.payload.get("old_text"): + raise SafetyRefusalError("the crontab changed after the plan was prepared; refusing to apply.") + elif plan.payload.get("noop_mode") == "file": + snapshots: dict[str, str | None] = plan.payload.get("snapshots", {}) + if any(_read_text(path) != snapshot for path, snapshot in snapshots.items()): + raise SafetyRefusalError( + "the system cron drop-in changed after the plan was prepared; refusing to apply." + ) + return MutationResult(changed=False) + def _apply_crontab(self, plan: Plan) -> MutationResult: old_text = plan.payload.get("old_text", "") new_text = plan.payload.get("new_text", "") @@ -925,6 +1125,10 @@ class CronBackend(SchedulerBackend): message = "System cron drop-in installed." elif action == "update": message = "System cron drop-in updated." + elif action == "enable": + message = "System cron job enabled." + elif action == "disable": + message = "System cron job disabled." elif action == "remove": message = "System cron drop-in removed." else: @@ -965,7 +1169,7 @@ class CronBackend(SchedulerBackend): action = plan.payload.get("action") expected_line = plan.payload.get("expected_line") if plan.payload.get("mode") == "file": - if action not in {"create", "update"} or not name: + if action not in {"create", "update", "enable", "disable"} or not name: return document = CrontabDocument(_read_text(plan.payload.get("path", "")) or "", system=True) try: @@ -976,7 +1180,7 @@ class CronBackend(SchedulerBackend): raise OperationalError("the drop-in was written but the schedls block could not be verified.") return document = self.read() - if action in {"create", "update"} and name: + if action in {"create", "update", "enable", "disable"} and name: try: job_line = document.managed_job_line(name) except SafetyRefusalError as exc: @@ -993,16 +1197,20 @@ class CronBackend(SchedulerBackend): if not isinstance(expected_line, str): return False expected_entries = [ - entry for entry in CrontabDocument(expected_line, system=system).entries if entry.kind == "job" + entry + for entry in CrontabDocument(expected_line, system=system).entries + if entry.kind in {"job", "disabled_job"} ] if len(expected_entries) != 1: return False expected = expected_entries[0] return ( + job_line.kind, job_line.expression, job_line.user, job_line.command, ) == ( + expected.kind, expected.expression, expected.user, expected.command, diff --git a/src/schedls/backends/systemd.py b/src/schedls/backends/systemd.py index 21223e4..a919d2f 100644 --- a/src/schedls/backends/systemd.py +++ b/src/schedls/backends/systemd.py @@ -7,6 +7,7 @@ import os import re import tempfile from collections.abc import Sequence +from dataclasses import replace from datetime import UTC, datetime from ..errors import ( @@ -390,6 +391,8 @@ class SystemdBackend(SchedulerBackend): if current.backend is not Backend.SYSTEMD or current.systemd is None: raise SafetyRefusalError(f"{current.name} is not a systemd timer.") self._require_scope(current.scope) + if not spec.replace_environment: + spec = replace(spec, environment=current.systemd.environment) self._validate_spec(spec) service_unit = unit_name(current.name, "service") timer_unit = unit_name(current.name, "timer") diff --git a/src/schedls/cli.py b/src/schedls/cli.py index d2d91d9..d029558 100644 --- a/src/schedls/cli.py +++ b/src/schedls/cli.py @@ -150,7 +150,12 @@ def build_parser() -> argparse.ArgumentParser: return parser -def _add_schedule_options(parser: argparse.ArgumentParser, *, creation: bool) -> None: +def _add_schedule_options( + parser: argparse.ArgumentParser, + *, + creation: bool, + environment_clearable: bool = False, +) -> None: parser.add_argument("--calendar", action="append", metavar="EXPR", help="native OnCalendar expression (repeatable)") parser.add_argument("--daily", metavar="TIME", help="daily at TIME (HH:MM)") parser.add_argument("--weekdays", metavar="TIME", help="Mon..Fri at TIME") @@ -161,7 +166,15 @@ def _add_schedule_options(parser: argparse.ArgumentParser, *, creation: bool) -> parser.add_argument("--jitter", metavar="DURATION", help="randomized delay, e.g. 5min (systemd timers)") parser.add_argument("--accuracy", metavar="DURATION", help="timer accuracy window, e.g. 1min (systemd timers)") parser.add_argument("--working-directory", metavar="PATH", help="run the command from PATH (systemd timers)") - parser.add_argument("--env", action="append", metavar="KEY=VALUE", help="set an environment variable (repeatable)") + environment_options = parser.add_mutually_exclusive_group() if environment_clearable else parser + environment_options.add_argument( + "--env", + action="append", + metavar="KEY=VALUE", + help="set an environment variable (repeatable)", + ) + if environment_clearable: + environment_options.add_argument("--clear-env", action="store_true", help="remove all environment variables") parser.add_argument("--shell", metavar="SCRIPT", help="run SCRIPT through /bin/sh instead of an argv") if creation: scope = parser.add_mutually_exclusive_group() @@ -184,7 +197,7 @@ def _add_creation_options(parser: argparse.ArgumentParser) -> None: def _add_edit_options(parser: argparse.ArgumentParser) -> None: - _add_schedule_options(parser, creation=False) + _add_schedule_options(parser, creation=False, environment_clearable=True) parser.add_argument( "--command", action="store_true", @@ -359,7 +372,7 @@ def _cmd_set_enabled(context: Context, args: argparse.Namespace, enabled: bool) if not (backend.capabilities.enable if enabled else backend.capabilities.disable): raise UsageError( f"{job.backend.value} does not support enable/disable.", - hint="Cron has no universal native enabled/disabled concept.", + hint="This scheduler backend does not expose an enabled/disabled state for this job.", ) plan = backend.plan_set_enabled(job, enabled) result = mutate_ops.run_plan( @@ -463,6 +476,17 @@ def _parse_environment(entries: list[str] | None) -> tuple[tuple[str, str], ...] return tuple(result) +def _environment_for_edit( + args: argparse.Namespace, + existing: tuple[tuple[str, str], ...], +) -> tuple[tuple[str, str], ...]: + if args.clear_env: + return () + if args.env is not None: + return _parse_environment(args.env) + return existing + + def _calendar_from_args(args: argparse.Namespace) -> list[str]: convenience_exprs: list[str] = [] if args.daily: @@ -526,8 +550,6 @@ def _spec_for_new(args: argparse.Namespace, tail: list[str]) -> JobSpec: raise UsageError("--calendar is a systemd option; use --cron-expr for cron.") if args.jitter or args.accuracy or args.persistent or args.working_directory: raise UsageError("--jitter/--accuracy/--persistent/--working-directory are systemd options.") - if args.env: - raise UsageError("--env is a systemd option; cron environment variables are not supported yet.") expression = _cron_expression_from_args(args) if not expression: raise UsageError("no schedule given.", hint="Use --cron-expr or a convenience flag such as --daily.") @@ -539,6 +561,7 @@ def _spec_for_new(args: argparse.Namespace, tail: list[str]) -> JobSpec: command=command, cron_expression=expression, run_as=args.run_as or "root", + environment=environment, ) if args.run_as: raise UsageError("--run-as requires --system for cron jobs.", hint="Cron jobs run as the current user.") @@ -548,6 +571,7 @@ def _spec_for_new(args: argparse.Namespace, tail: list[str]) -> JobSpec: scope=Scope.USER, command=command, cron_expression=expression, + environment=environment, ) @@ -581,15 +605,14 @@ def _spec_for_edit(job: ScheduledJob, args: argparse.Namespace, tail: list[str]) working_directory=args.working_directory if args.working_directory is not None else (job.systemd.working_directory if job.systemd else None), - environment=_parse_environment(args.env) if args.env else (job.systemd.environment if job.systemd else ()), + environment=_environment_for_edit(args, job.systemd.environment if job.systemd else ()), + replace_environment=args.clear_env or args.env is not None, ) if job.backend is Backend.CRON: if args.calendar: raise UsageError("--calendar is a systemd option; use --cron-expr for cron.") if args.persistent or args.no_persistent or args.jitter or args.accuracy or args.working_directory: raise UsageError("--persistent/--jitter/--accuracy/--working-directory are systemd options.") - if args.env: - raise UsageError("--env is a systemd option; cron environment variables are not supported yet.") expression = _cron_expression_from_args(args) if expression is None: expression = job.cron.expression if job.cron else job.schedule.expression @@ -601,6 +624,8 @@ def _spec_for_edit(job: ScheduledJob, args: argparse.Namespace, tail: list[str]) scope=job.scope, command=command, cron_expression=expression, + environment=_environment_for_edit(args, job.cron.environment if job.cron else ()), + replace_environment=args.clear_env or args.env is not None, ) raise UsageError(f"editing {job.backend.value} jobs is not supported yet.") @@ -682,6 +707,7 @@ def _wizard_new(args: argparse.Namespace, tail: list[str], prompter: Prompter) - _apply_schedule(args, prompter.schedule(backend)) if backend is Backend.SYSTEMD: _wizard_advanced(args, prompter) + _wizard_new_environment(args, prompter) return args, tail @@ -705,10 +731,13 @@ def _wizard_advanced(args: argparse.Namespace, prompter: Prompter) -> None: working_directory = prompter.optional_text("Working directory (absolute path)") if working_directory is not None: args.working_directory = working_directory + + +def _wizard_new_environment(args: argparse.Namespace, prompter: Prompter) -> None: + if args.env is not None: + return if prompter.yes_no("Add environment variables?", default=False): - entries = prompter.environment() - if entries: - args.env = entries + args.env = prompter.environment() def _wizard_edit( @@ -735,6 +764,8 @@ def _wizard_edit( _apply_schedule(args, prompter.schedule(job.backend)) if job.backend is Backend.SYSTEMD: _wizard_edit_advanced(job, args, prompter) + if job.backend in {Backend.SYSTEMD, Backend.CRON}: + _wizard_edit_environment(args, prompter) return args, tail @@ -760,8 +791,18 @@ def _wizard_edit_advanced(job: ScheduledJob, args: argparse.Namespace, prompter: ) if working_directory is not None: args.working_directory = working_directory - if prompter.yes_no("Replace environment variables?", default=False): - args.env = prompter.environment() + + +def _wizard_edit_environment(args: argparse.Namespace, prompter: Prompter) -> None: + if args.env is not None or args.clear_env: + return + if not prompter.yes_no("Replace environment variables?", default=False): + return + entries = prompter.environment() + if entries: + args.env = entries + else: + args.clear_env = True def _report_created(context: Context, spec: JobSpec) -> None: diff --git a/src/schedls/models.py b/src/schedls/models.py index 9fe3b87..bdc1f46 100644 --- a/src/schedls/models.py +++ b/src/schedls/models.py @@ -141,6 +141,7 @@ class JobSpec: accuracy: str | None = None working_directory: str | None = None environment: tuple[tuple[str, str], ...] = field(default_factory=tuple) + replace_environment: bool = False def schedule(self) -> Schedule: if self.backend is Backend.SYSTEMD: diff --git a/src/schedls/operations/inspect.py b/src/schedls/operations/inspect.py index e7940d5..44f86f7 100644 --- a/src/schedls/operations/inspect.py +++ b/src/schedls/operations/inspect.py @@ -23,6 +23,7 @@ def collect(backends: Sequence[SchedulerBackend], scopes: Sequence[Scope]) -> tu continue try: jobs.extend(backend.discover(scopes)) + warnings.extend(f"{backend.name} discovery warning: {warning}" for warning in backend.discovery_warnings()) except SchedlsError as exc: warnings.append(f"{backend.name} discovery failed: {exc.message}") jobs.sort(key=lambda job: (job.name, job.backend.value)) @@ -142,6 +143,8 @@ def render_show(output: Output, job: ScheduledJob) -> None: details.append(("SHELL", job.cron.shell)) if job.cron.mailto: details.append(("MAILTO", job.cron.mailto)) + for key, value in job.cron.environment: + details.append(("Environment", f"{key}={value}")) width = max(len(key) for key, _ in [*core, *details]) output.key_values(core, width=width) if details: diff --git a/src/schedls/renderers/cron.py b/src/schedls/renderers/cron.py index 66d1abf..2800a45 100644 --- a/src/schedls/renderers/cron.py +++ b/src/schedls/renderers/cron.py @@ -10,6 +10,7 @@ from __future__ import annotations import re import shlex +from dataclasses import dataclass from ..errors import InvalidScheduleError, SafetyRefusalError from ..models import Command @@ -17,6 +18,7 @@ from ..security import MANAGED_COMMENT, has_unsafe_control_characters BLOCK_BEGIN = "# schedls:begin name={name}{extra}" BLOCK_END = "# schedls:end name={name}" +DISABLED_PREFIX = "# schedls:disabled " _MARKER_RE = re.compile(r"^#\s*schedls:(?Pbegin|end)\s+(?P.*)$") _FIELD_RE = re.compile(r"^[0-9A-Za-z*,/\-]+$") @@ -33,6 +35,15 @@ _NICKNAMES = { _RANGE_RE = re.compile(r"^[0-9]{1,2}-[0-9]{1,2}$") _SYSTEM_NAME_RE = re.compile(r"\A[A-Za-z0-9][A-Za-z0-9_-]{0,63}\Z") _USER_RE = re.compile(r"\A[A-Za-z0-9_][A-Za-z0-9_.-]{0,31}\Z") +_ENV_NAME_RE = re.compile(r"\A[A-Za-z_][A-Za-z0-9_]*\Z") +_ENV_ASSIGNMENT_START_RE = re.compile(r"\A[A-Za-z_][A-Za-z0-9_]*=") + + +@dataclass(frozen=True) +class EnvironmentPrefix: + environment: tuple[tuple[str, str], ...] + command: str + malformed: bool = False def escape_percent(text: str) -> str: @@ -46,19 +57,74 @@ def escape_percent(text: str) -> str: return text.replace("%", "\\%") -def render_command(command: Command) -> str: +def unescape_percent(text: str) -> str: + """Model cron's conversion of escaped percent characters before shell parsing.""" + return text.replace("\\%", "%") + + +def render_environment(environment: tuple[tuple[str, str], ...]) -> str: + assignments: list[str] = [] + for key, value in environment: + if not _ENV_NAME_RE.match(key): + raise InvalidScheduleError(f"invalid environment variable name: {key!r}") + if has_unsafe_control_characters(value): + raise SafetyRefusalError("cron environment values must not contain NUL or newline") + assignments.append(f"{key}={shlex.quote(value)}") + return " ".join(assignments) + + +def parse_environment_prefix(command: str) -> EnvironmentPrefix: + """Parse schedls's canonical leading shell assignments without executing them.""" + lexer = shlex.shlex(unescape_percent(command), posix=True) + lexer.whitespace_split = True + lexer.commenters = "" + environment: list[tuple[str, str]] = [] + try: + while True: + token = lexer.get_token() + if token is None or token == lexer.eof: + break + key, separator, value = token.partition("=") + if not separator or not _ENV_NAME_RE.match(key): + break + environment.append((key, value)) + except ValueError: + if environment or _ENV_ASSIGNMENT_START_RE.match(command): + return EnvironmentPrefix((), command, malformed=True) + return EnvironmentPrefix((), command) + + if not environment: + return EnvironmentPrefix((), command) + try: + rendered_prefix = escape_percent(render_environment(tuple(environment))) + except (InvalidScheduleError, SafetyRefusalError): + return EnvironmentPrefix((), command, malformed=True) + if not command.startswith(rendered_prefix): + return EnvironmentPrefix((), command, malformed=True) + remainder = command[len(rendered_prefix) :] + if not remainder.startswith(" ") or not remainder[1:]: + return EnvironmentPrefix((), command, malformed=True) + return EnvironmentPrefix(tuple(environment), remainder[1:]) + + +def _render_command_text(command: Command) -> str: if command.shell: raw = command.raw or "" if has_unsafe_control_characters(raw): raise SafetyRefusalError("shell command must not contain NUL or newline") - return escape_percent(raw) + return raw if not command.argv: raise InvalidScheduleError("no command given") for arg in command.argv: if has_unsafe_control_characters(arg): raise SafetyRefusalError("cron command arguments must not contain NUL or newline") - quoted = " ".join(shlex.quote(arg) for arg in command.argv) - return escape_percent(quoted) + return " ".join(shlex.quote(arg) for arg in command.argv) + + +def render_command(command: Command, *, environment: tuple[tuple[str, str], ...] = ()) -> str: + command_text = _render_command_text(command) + assignments = render_environment(environment) + return escape_percent(f"{assignments} {command_text}" if assignments else command_text) def _validate_field(field: str, *, low: int, high: int, names: bool) -> None: @@ -120,8 +186,51 @@ def validate_expression(expression: str) -> str: return " ".join(fields) -def render_line(expression: str, command: Command) -> str: - return f"{validate_expression(expression)} {render_command(command)}" +def render_line( + expression: str, + command: Command, + *, + environment: tuple[tuple[str, str], ...] = (), +) -> str: + return f"{validate_expression(expression)} {render_command(command, environment=environment)}" + + +def render_line_from_raw_command( + expression: str, + raw_command: str, + *, + environment: tuple[tuple[str, str], ...] = (), + user: str | None = None, +) -> str: + """Render an updated line while preserving an existing cron command field. + + ``raw_command`` is already cron-escaped and is therefore appended unchanged. + """ + if not raw_command or has_unsafe_control_characters(raw_command): + raise SafetyRefusalError("existing cron command cannot be safely preserved.") + command_field = raw_command + assignments = render_environment(environment) + if assignments: + command_field = f"{escape_percent(assignments)} {raw_command}" + fields = [validate_expression(expression)] + if user is not None: + fields.append(validate_run_as(user)) + fields.append(command_field) + return " ".join(fields) + + +def render_disabled_line(line: str) -> str: + if has_unsafe_control_characters(line): + raise SafetyRefusalError("cron job line must not contain NUL or newline") + if line.startswith(DISABLED_PREFIX): + return line + return f"{DISABLED_PREFIX}{line}" + + +def render_enabled_line(line: str) -> str: + if not line.startswith(DISABLED_PREFIX): + raise SafetyRefusalError("cron job line is not disabled by schedls") + return line[len(DISABLED_PREFIX) :] def validate_run_as(user: str) -> str: @@ -147,8 +256,16 @@ def validate_system_name(name: str) -> str: return name -def render_system_line(expression: str, command: Command, user: str) -> str: - return f"{validate_expression(expression)} {validate_run_as(user)} {render_command(command)}" +def render_system_line( + expression: str, + command: Command, + user: str, + *, + environment: tuple[tuple[str, str], ...] = (), +) -> str: + return ( + f"{validate_expression(expression)} {validate_run_as(user)} {render_command(command, environment=environment)}" + ) def render_block(name: str, lines: list[str], *, extra: str = "") -> str: diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 513487c..c0fa34c 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -1,14 +1,31 @@ from __future__ import annotations +import argparse +import io import os import subprocess import sys import pytest -from schedls.cli import _parse_environment, _spec_for_edit, _spec_for_new, build_parser, main, split_command +from schedls.backends.cron import CronBackend +from schedls.backends.systemd import SystemdBackend +from schedls.cli import ( + Context, + _cmd_set_enabled, + _parse_environment, + _spec_for_edit, + _spec_for_new, + build_parser, + main, + split_command, +) from schedls.errors import InvalidNameError, UsageError +from schedls.interact import Interaction from schedls.models import Backend, Command, CronDetails, JobSource, Schedule, ScheduledJob, ScheduleKind, Scope +from schedls.output import Output + +from ..conftest import FakeRunner def test_split_command() -> None: @@ -86,6 +103,16 @@ def test_spec_for_new_cron() -> None: assert spec.cron_expression == "0 4 * * 0" +def test_spec_for_new_cron_accepts_environment() -> None: + args = build_parser().parse_args( + ["new", "cleanup", "--cron", "--cron-expr", "0 4 * * 0", "--env", "GREETING=hello world", "--env", "RATE=100%"] + ) + + spec = _spec_for_new(args, ["/usr/local/bin/cleanup"]) + + assert spec.environment == (("GREETING", "hello world"), ("RATE", "100%")) + + def test_spec_for_new_system_scope() -> None: args = build_parser().parse_args(["new", "x", "--timer", "--daily", "02:00", "--system"]) spec = _spec_for_new(args, ["/bin/true"]) @@ -137,7 +164,7 @@ def test_spec_for_new_shell_mode() -> None: def test_spec_for_new_rejects_systemd_flags_for_cron() -> None: - for extra in (["--env", "A=1"], ["--working-directory", "/srv/data"], ["--persistent"], ["--jitter", "5min"]): + for extra in (["--working-directory", "/srv/data"], ["--persistent"], ["--jitter", "5min"]): args = build_parser().parse_args(["new", "x", "--cron", "--daily", "02:00", *extra]) with pytest.raises(UsageError): _spec_for_new(args, ["/bin/true"]) @@ -174,7 +201,7 @@ def test_spec_for_new_rejects_run_as_for_timer() -> None: _spec_for_new(args, ["/bin/true"]) -def _cron_job() -> ScheduledJob: +def _cron_job(*, environment: tuple[tuple[str, str], ...] = ()) -> ScheduledJob: return ScheduledJob( name="cleanup", backend=Backend.CRON, @@ -184,19 +211,21 @@ def _cron_job() -> ScheduledJob: schedule=Schedule(ScheduleKind.CRON, "0 2 * * *"), command=Command(raw="/usr/local/bin/cleanup"), source=JobSource("current user's crontab", line=1), - cron=CronDetails(expression="0 2 * * *", line=1), + cron=CronDetails(expression="0 2 * * *", line=1, environment=environment), ) def test_spec_for_edit_cron_preserves_command_and_changes_schedule() -> None: args = build_parser().parse_args(["edit", "cleanup", "--daily", "03:00"]) - job = _cron_job() + job = _cron_job(environment=(("OLD", "value"),)) spec = _spec_for_edit(job, args, []) assert spec.backend is Backend.CRON assert spec.cron_expression == "0 3 * * *" assert spec.command == job.command + assert spec.environment == (("OLD", "value"),) + assert spec.replace_environment is False def test_spec_for_edit_cron_replaces_command() -> None: @@ -207,7 +236,61 @@ def test_spec_for_edit_cron_replaces_command() -> None: assert spec.command.argv == ("/usr/local/bin/cleanup", "--full") -@pytest.mark.parametrize("extra", [["--calendar", "daily"], ["--persistent"], ["--env", "A=1"]]) +def test_spec_for_edit_cron_replaces_environment() -> None: + args = build_parser().parse_args(["edit", "cleanup", "--env", "NEW=value", "--env", "EMPTY="]) + + spec = _spec_for_edit(_cron_job(environment=(("OLD", "value"),)), args, []) + + assert spec.environment == (("NEW", "value"), ("EMPTY", "")) + assert spec.replace_environment is True + + +def test_spec_for_edit_cron_clears_environment() -> None: + args = build_parser().parse_args(["edit", "cleanup", "--clear-env"]) + + spec = _spec_for_edit(_cron_job(environment=(("OLD", "value"),)), args, []) + + assert spec.environment == () + assert spec.replace_environment is True + + +def test_edit_environment_options_are_mutually_exclusive() -> None: + with pytest.raises(SystemExit): + build_parser().parse_args(["edit", "cleanup", "--env", "NEW=value", "--clear-env"]) + + +def test_disable_managed_cron_job_through_cli() -> None: + state = {"text": ("# schedls:begin name=backup\n0 2 * * * /usr/bin/backup\n# schedls:end name=backup\n")} + + def crontab(argv, input_text): + if argv == ["crontab", "-l"]: + return state["text"], "" + if argv == ["crontab", "-"]: + state["text"] = input_text or "" + return "", "" + return 1, "", "unsupported" + + runner = FakeRunner(handlers={"crontab": crontab}, available={"crontab"}) + stream = io.StringIO() + output = Output(color="never", stdout=stream) + context = Context( + runner=runner, + output=output, + interaction=Interaction(output=output, assume_yes=True), + systemd=SystemdBackend(runner), + cron=CronBackend(runner), + ) + dry_run_args = argparse.Namespace(name="backup", dry_run=True, show_files=False) + args = argparse.Namespace(name="backup", dry_run=False, show_files=False) + + assert _cmd_set_enabled(context, dry_run_args, False) == 0 + assert state["text"] == ("# schedls:begin name=backup\n0 2 * * * /usr/bin/backup\n# schedls:end name=backup\n") + assert _cmd_set_enabled(context, args, False) == 0 + assert "# schedls:disabled 0 2 * * * /usr/bin/backup" in state["text"] + assert "Disabled backup." in stream.getvalue() + + +@pytest.mark.parametrize("extra", [["--calendar", "daily"], ["--persistent"]]) def test_spec_for_edit_cron_rejects_systemd_options(extra: list[str]) -> None: args = build_parser().parse_args(["edit", "cleanup", *extra]) diff --git a/tests/unit/test_cron_system.py b/tests/unit/test_cron_system.py index e400ea6..6845c23 100644 --- a/tests/unit/test_cron_system.py +++ b/tests/unit/test_cron_system.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +from dataclasses import replace import pytest @@ -8,6 +9,8 @@ from schedls.backends.base import FileChange, Plan from schedls.backends.cron import CronBackend, CronPaths, CrontabDocument from schedls.errors import ConflictError, InvalidScheduleError, SafetyRefusalError from schedls.models import Backend, Command, JobSpec, Scope +from schedls.operations import inspect as inspect_ops +from schedls.output import job_to_dict from schedls.renderers import cron as renderer @@ -49,7 +52,7 @@ def test_discovers_system_crontabs(tmp_path, fake_runner) -> None: _write(tmp_path / "etc/cron.d/backup", "0 2 * * * root /usr/bin/backup\n") _write( tmp_path / "etc/cron.d/schedls-cleanup", - renderer.render_system_file("cleanup", ["30 3 * * 0 deploy /usr/bin/cleanup"]), + renderer.render_system_file("cleanup", ["30 3 * * 0 deploy GREETING='hello world' /usr/bin/cleanup"]), ) _write(tmp_path / "etc/cron.d/with.dot", "0 2 * * * root /usr/bin/ignored\n") _write(tmp_path / "etc/cron.d/.placeholder", "") @@ -70,9 +73,353 @@ def test_discovers_system_crontabs(tmp_path, fake_runner) -> None: assert managed.managed is True assert managed.cron.user == "deploy" assert managed.schedule.expression == "30 3 * * 0" + assert managed.command.raw == "/usr/bin/cleanup" + assert managed.cron.environment == (("GREETING", "hello world"),) assert managed.source.path == paths.cron_d + "/schedls-cleanup" +def test_discovers_disabled_managed_user_cron_job(tmp_path, fake_runner) -> None: + text = ( + "# schedls:begin name=backup\n" + "# schedls:disabled 0 2 * * * /usr/bin/backup /srv/data\n" + "# schedls:end name=backup\n" + ) + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return text, "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + + jobs = CronBackend(fake_runner, _make_paths(tmp_path)).discover([Scope.USER]) + + assert len(jobs) == 1 + job = jobs[0] + assert job.managed is True + assert job.enabled is False + assert job.schedule.expression == "0 2 * * *" + assert job.command.raw == "/usr/bin/backup /srv/data" + assert inspect_ops.filter_jobs(jobs, enabled=False) == jobs + assert inspect_ops.job_status(job) == "disabled" + assert job_to_dict(job)["enabled"] is False + + +def test_discovers_managed_cron_environment_as_logical_values(tmp_path, fake_runner) -> None: + text = ( + "# schedls:begin name=backup\n" + "0 2 * * * GREETING='hello world' RATE=100\\% /usr/bin/backup /srv/data\n" + "# schedls:end name=backup\n" + ) + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return text, "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + + job = CronBackend(fake_runner, _make_paths(tmp_path)).discover([Scope.USER])[0] + + assert job.command.raw == "/usr/bin/backup /srv/data" + assert job.cron is not None + assert job.cron.environment == (("GREETING", "hello world"), ("RATE", "100%")) + assert job_to_dict(job)["cron"]["environment"] == [["GREETING", "hello world"], ["RATE", "100%"]] + + +def test_collect_warns_about_malformed_disabled_marker(tmp_path, fake_runner) -> None: + text = "# schedls:begin name=backup\n# schedls:disabled not a cron job\n# schedls:end name=backup\n" + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return text, "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + + _, warnings = inspect_ops.collect([CronBackend(fake_runner, _make_paths(tmp_path))], [Scope.USER]) + + assert warnings == ["cron discovery warning: current user's crontab: line 2: malformed schedls disabled job"] + + +def test_collect_warns_and_mutation_refuses_noncanonical_environment_prefix(tmp_path, fake_runner) -> None: + text = '# schedls:begin name=backup\n0 2 * * * GREETING="hello world" /usr/bin/backup\n# schedls:end name=backup\n' + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return text, "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + backend = CronBackend(fake_runner, _make_paths(tmp_path)) + + jobs, warnings = inspect_ops.collect([backend], [Scope.USER]) + + assert warnings == ["cron discovery warning: current user's crontab: line 2: malformed schedls environment prefix"] + with pytest.raises(SafetyRefusalError): + backend.plan_update( + jobs[0], + JobSpec( + name="backup", + backend=Backend.CRON, + scope=Scope.USER, + command=jobs[0].command, + cron_expression="30 3 * * *", + ), + ) + + +def test_disabled_marker_outside_managed_block_is_not_discovered_as_a_job(tmp_path, fake_runner) -> None: + text = "# schedls:disabled 0 2 * * * /usr/bin/backup\n" + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return text, "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + + jobs = CronBackend(fake_runner, _make_paths(tmp_path)).discover([Scope.USER]) + + assert jobs == [] + + +def test_plan_set_enabled_user_cron_round_trips_job_line(tmp_path, fake_runner) -> None: + original = ( + "# unrelated comment\n" + "# schedls:begin name=backup\n" + "0 2 * * * /usr/bin/backup /srv/data\n" + "# schedls:end name=backup\n" + ) + state = {"text": original} + + def crontab(argv, input_text): + if argv == ["crontab", "-l"]: + return state["text"], "" + if argv == ["crontab", "-"]: + state["text"] = input_text or "" + return "", "" + return 1, "", "unsupported" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + backend = CronBackend(fake_runner, _make_paths(tmp_path)) + job = backend.discover([Scope.USER])[0] + + disabled = backend.plan_set_enabled(job, False) + + assert disabled.action == "disable" + assert "# unrelated comment" in disabled.payload["new_text"] + assert "# schedls:disabled 0 2 * * * /usr/bin/backup /srv/data" in disabled.payload["new_text"] + assert backend.apply(disabled).changed is True + assert backend.discover([Scope.USER])[0].enabled is False + + enabled = backend.plan_set_enabled(backend.discover([Scope.USER])[0], True) + + assert enabled.action == "enable" + assert backend.apply(enabled).changed is True + assert state["text"] == original + + +def test_plan_set_enabled_is_noop_when_cron_job_already_has_requested_state(tmp_path, fake_runner) -> None: + text = "# schedls:begin name=backup\n0 2 * * * /usr/bin/backup\n# schedls:end name=backup\n" + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return text, "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + backend = CronBackend(fake_runner, _make_paths(tmp_path)) + job = backend.discover([Scope.USER])[0] + + plan = backend.plan_set_enabled(job, True) + result = backend.apply(plan) + + assert plan.payload["mode"] == "noop" + assert result.changed is False + assert [call[0] for call in fake_runner.calls] == [ + ["crontab", "-l"], + ["crontab", "-l"], + ["crontab", "-l"], + ] + + +def test_plan_set_enabled_rechecks_crontab_before_treating_request_as_a_noop(tmp_path, fake_runner) -> None: + active = "# schedls:begin name=backup\n0 2 * * * /usr/bin/backup\n# schedls:end name=backup\n" + disabled = active.replace("0 2 * * *", "# schedls:disabled 0 2 * * *") + state = {"text": active} + + def crontab(argv, input_text): + if argv == ["crontab", "-l"]: + return state["text"], "" + if argv == ["crontab", "-"]: + state["text"] = input_text or "" + return "", "" + return 1, "", "unsupported" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + backend = CronBackend(fake_runner, _make_paths(tmp_path)) + stale_job = backend.discover([Scope.USER])[0] + state["text"] = disabled + + plan = backend.plan_set_enabled(stale_job, True) + + assert plan.payload["mode"] == "crontab" + assert backend.apply(plan).changed is True + assert state["text"] == active + + +def test_noop_set_enabled_refuses_if_crontab_changes_before_apply(tmp_path, fake_runner) -> None: + active = "# schedls:begin name=backup\n0 2 * * * /usr/bin/backup\n# schedls:end name=backup\n" + state = {"text": active} + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return state["text"], "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + backend = CronBackend(fake_runner, _make_paths(tmp_path)) + plan = backend.plan_set_enabled(backend.discover([Scope.USER])[0], True) + state["text"] = active.replace("0 2 * * *", "# schedls:disabled 0 2 * * *") + + with pytest.raises(SafetyRefusalError, match="crontab changed"): + backend.apply(plan) + + +def test_noop_set_enabled_refuses_if_system_drop_in_changes_before_apply(tmp_path, fake_runner, monkeypatch) -> None: + paths = _make_paths(tmp_path) + path = _write( + tmp_path / "etc/cron.d/schedls-backup", + renderer.render_system_file("backup", ["0 2 * * * root /usr/bin/backup"]), + ) + backend = CronBackend(fake_runner, paths) + job = backend.discover([Scope.SYSTEM])[0] + monkeypatch.setattr(os, "geteuid", lambda: 0) + monkeypatch.setattr("schedls.backends.cron.check_trusted_directory", lambda path, *, expected_uid: None) + + plan = backend.plan_set_enabled(job, True) + path.write_text(renderer.render_system_file("backup", ["# schedls:disabled 0 2 * * * root /usr/bin/backup"])) + + with pytest.raises(SafetyRefusalError, match="drop-in changed"): + backend.apply(plan) + + +def test_plan_set_enabled_system_cron_preserves_run_as(tmp_path, fake_runner, monkeypatch) -> None: + paths = _make_paths(tmp_path) + path = _write( + tmp_path / "etc/cron.d/schedls-backup", + renderer.render_system_file("backup", ["0 2 * * * www-data /usr/bin/backup /srv/data"]), + ) + backend = CronBackend(fake_runner, paths) + job = backend.discover([Scope.SYSTEM])[0] + monkeypatch.setattr(os, "geteuid", lambda: 0) + trusted_directories: list[tuple[str, int]] = [] + + def check_trusted_directory(path: str, *, expected_uid: int) -> None: + trusted_directories.append((path, expected_uid)) + + monkeypatch.setattr("schedls.backends.cron.check_trusted_directory", check_trusted_directory) + + plan = backend.plan_set_enabled(job, False) + + assert plan.action == "disable" + assert plan.files[0].path == str(path) + assert "# schedls:disabled 0 2 * * * www-data /usr/bin/backup /srv/data" in plan.files[0].content + assert trusted_directories == [(paths.cron_d, 0)] + assert CronBackend(fake_runner, paths).capabilities.enable is True + assert CronBackend(fake_runner, paths).capabilities.disable is True + + +def test_plan_set_enabled_system_cron_requires_root(tmp_path, fake_runner, monkeypatch) -> None: + paths = _make_paths(tmp_path) + _write( + tmp_path / "etc/cron.d/schedls-backup", + renderer.render_system_file("backup", ["0 2 * * * root /usr/bin/backup"]), + ) + backend = CronBackend(fake_runner, paths) + job = backend.discover([Scope.SYSTEM])[0] + monkeypatch.setattr(os, "geteuid", lambda: 1000) + + with pytest.raises(SafetyRefusalError): + backend.plan_set_enabled(job, False) + + +def test_plan_set_enabled_system_cron_refuses_unexpected_path(tmp_path, fake_runner, monkeypatch) -> None: + paths = _make_paths(tmp_path) + _write( + tmp_path / "etc/cron.d/schedls-backup", + renderer.render_system_file("backup", ["0 2 * * * root /usr/bin/backup"]), + ) + backend = CronBackend(fake_runner, paths) + job = backend.discover([Scope.SYSTEM])[0] + unexpected_job = replace(job, source=replace(job.source, path=str(tmp_path / "unexpected"))) + monkeypatch.setattr(os, "geteuid", lambda: 0) + monkeypatch.setattr("schedls.backends.cron.check_trusted_directory", lambda path, *, expected_uid: None) + + with pytest.raises(SafetyRefusalError, match="unexpected system cron file"): + backend.plan_set_enabled(unexpected_job, False) + + +def test_plan_update_disabled_system_cron_preserves_disabled_state(tmp_path, fake_runner, monkeypatch) -> None: + paths = _make_paths(tmp_path) + _write( + tmp_path / "etc/cron.d/schedls-backup", + renderer.render_system_file("backup", ["# schedls:disabled 0 2 * * * www-data /usr/bin/backup"]), + ) + backend = CronBackend(fake_runner, paths) + job = backend.discover([Scope.SYSTEM])[0] + monkeypatch.setattr(os, "geteuid", lambda: 0) + monkeypatch.setattr("schedls.backends.cron.check_trusted_directory", lambda path, *, expected_uid: None) + + plan = backend.plan_update( + job, + _system_spec(name="backup", command=job.command, cron_expression="30 3 * * *"), + ) + + assert "# schedls:disabled 30 3 * * * www-data /usr/bin/backup" in plan.files[0].content + + +def test_plan_update_disabled_user_cron_preserves_disabled_state(tmp_path, fake_runner) -> None: + text = ( + "# schedls:begin name=backup\n" + "# schedls:disabled 0 2 * * * /usr/bin/backup /srv/data\n" + "# schedls:end name=backup\n" + ) + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return text, "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + backend = CronBackend(fake_runner, _make_paths(tmp_path)) + job = backend.discover([Scope.USER])[0] + + plan = backend.plan_update( + job, + JobSpec( + name="backup", + backend=Backend.CRON, + scope=Scope.USER, + command=job.command, + cron_expression="30 3 * * *", + ), + ) + + assert "# schedls:disabled 30 3 * * * /usr/bin/backup /srv/data" in plan.payload["new_text"] + + def test_discovers_periodic_directories(tmp_path, fake_runner) -> None: paths = _make_paths(tmp_path) _write(tmp_path / "etc/cron.daily/logrotate", "#!/bin/sh\n", mode=0o755) @@ -128,6 +475,20 @@ def test_malformed_drop_in_is_unmanaged(tmp_path, fake_runner) -> None: assert jobs[0].warnings +def test_collect_warns_when_malformed_drop_in_has_no_job_lines(tmp_path, fake_runner, monkeypatch) -> None: + paths = _make_paths(tmp_path) + path = _write(tmp_path / "etc/cron.d/schedls-broken", "") + monkeypatch.setattr(os, "geteuid", lambda: 0) + + jobs, warnings = inspect_ops.collect([CronBackend(fake_runner, paths)], [Scope.SYSTEM]) + + assert jobs == [] + assert warnings == [ + f"cron discovery warning: {path} looks like a schedls drop-in but its managed block is missing " + "or malformed; treating its jobs as unmanaged." + ] + + def _system_spec(**overrides) -> JobSpec: values = { "name": "backup", @@ -141,6 +502,58 @@ def _system_spec(**overrides) -> JobSpec: return JobSpec(**values) +def test_plan_create_user_cron_with_execution_local_environment(tmp_path, fake_runner) -> None: + state = {"text": "# unrelated job\n0 1 * * * /usr/bin/unrelated\n"} + + def crontab(argv, input_text): + if argv == ["crontab", "-l"]: + assert input_text is None + return state["text"], "" + if argv == ["crontab", "-"]: + state["text"] = input_text or "" + return "", "" + return 1, "", "unsupported" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + backend = CronBackend(fake_runner, _make_paths(tmp_path)) + plan = backend.plan_create( + JobSpec( + name="backup", + backend=Backend.CRON, + scope=Scope.USER, + command=Command(argv=("/usr/bin/backup", "/srv/data")), + cron_expression="0 2 * * *", + environment=(("GREETING", "hello world"), ("RATE", "100%"), ("MESSAGE", "50% off")), + ) + ) + + assert "0 1 * * * /usr/bin/unrelated" in plan.payload["new_text"] + assert ( + "0 2 * * * GREETING='hello world' RATE=100\\% MESSAGE='50\\% off' /usr/bin/backup /srv/data" + in plan.payload["new_text"] + ) + assert "GREETING=" not in plan.payload["new_text"].split("# schedls:begin")[0] + assert backend.apply(plan).changed is True + jobs_by_name = {job.name: job for job in backend.discover([Scope.USER])} + backup = jobs_by_name["backup"] + assert backup.cron.environment == (("GREETING", "hello world"), ("RATE", "100%"), ("MESSAGE", "50% off")) + update_plan = backend.plan_update( + backup, + JobSpec( + name="backup", + backend=Backend.CRON, + scope=Scope.USER, + command=backup.command, + cron_expression="30 3 * * *", + ), + ) + assert ( + "30 3 * * * GREETING='hello world' RATE=100\\% MESSAGE='50\\% off' /usr/bin/backup /srv/data" + in update_plan.payload["new_text"] + ) + + def test_plan_create_system_cron(tmp_path, fake_runner, monkeypatch) -> None: paths = _make_paths(tmp_path) (tmp_path / "etc/cron.d").mkdir(parents=True, exist_ok=True) @@ -158,6 +571,20 @@ def test_plan_create_system_cron(tmp_path, fake_runner, monkeypatch) -> None: assert ("Run as", "www-data") in plan.summary +def test_plan_create_system_cron_with_execution_local_environment(tmp_path, fake_runner, monkeypatch) -> None: + paths = _make_paths(tmp_path) + (tmp_path / "etc/cron.d").mkdir(parents=True, exist_ok=True) + monkeypatch.setattr(os, "geteuid", lambda: 0) + monkeypatch.setattr("schedls.backends.cron.check_trusted_directory", lambda path, *, expected_uid: None) + + plan = CronBackend(fake_runner, paths).plan_create( + _system_spec(environment=(("GREETING", "hello world"), ("RATE", "100%"))) + ) + + assert "0 2 * * * www-data GREETING='hello world' RATE=100\\% /usr/bin/backup /srv/data" in plan.files[0].content + assert "\nGREETING=" not in plan.files[0].content + + def test_plan_create_system_cron_requires_root(tmp_path, fake_runner, monkeypatch) -> None: paths = _make_paths(tmp_path) monkeypatch.setattr(os, "geteuid", lambda: 1000) @@ -185,7 +612,7 @@ def test_update_user_cron_preserves_existing_command(tmp_path, fake_runner) -> N document = CrontabDocument( "# unrelated comment\n" "# schedls:begin name=backup\n" - "0 2 * * * /usr/bin/printf '50\\%s\\n'\n" + "0 2 * * * GREETING='hello world' /usr/bin/printf '50\\%s\\n'\n" "# schedls:end name=backup\n" ) state = {"text": document.text} @@ -216,7 +643,7 @@ def test_update_user_cron_preserves_existing_command(tmp_path, fake_runner) -> N assert plan.action == "update" assert "# unrelated comment" in plan.payload["new_text"] - assert "30 3 * * * /usr/bin/printf '50\\%s\\n'" in plan.payload["new_text"] + assert "30 3 * * * GREETING='hello world' /usr/bin/printf '50\\%s\\n'" in plan.payload["new_text"] assert "50\\\\%" not in plan.payload["new_text"] state["text"] = state["text"].replace("0 2 * * *", "15 2 * * *") with pytest.raises(SafetyRefusalError, match="crontab changed"): @@ -237,6 +664,78 @@ def test_update_user_cron_preserves_existing_command(tmp_path, fake_runner) -> N assert state["text"] == plan.payload["new_text"] +def test_update_disabled_user_cron_replaces_environment_without_reescaping_command(tmp_path, fake_runner) -> None: + state = { + "text": ( + "# schedls:begin name=backup\n" + "# schedls:disabled 0 2 * * * OLD=old /usr/bin/printf '50\\%s\\n'\n" + "# schedls:end name=backup\n" + ) + } + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return state["text"], "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + backend = CronBackend(fake_runner, _make_paths(tmp_path)) + job = backend.discover([Scope.USER])[0] + + plan = backend.plan_update( + job, + JobSpec( + name="backup", + backend=Backend.CRON, + scope=Scope.USER, + command=job.command, + cron_expression="0 2 * * *", + environment=(("MESSAGE", "50% off"),), + replace_environment=True, + ), + ) + + assert "# schedls:disabled 0 2 * * * MESSAGE='50\\% off' /usr/bin/printf '50\\%s\\n'" in plan.payload["new_text"] + assert "OLD=old" not in plan.payload["new_text"] + assert "50\\\\%" not in plan.payload["new_text"] + + +def test_update_user_cron_clears_environment_without_reescaping_command(tmp_path, fake_runner) -> None: + state = { + "text": ( + "# schedls:begin name=backup\n0 2 * * * OLD=old /usr/bin/printf '50\\%s\\n'\n# schedls:end name=backup\n" + ) + } + + def crontab(argv, input_text): + assert argv == ["crontab", "-l"] + assert input_text is None + return state["text"], "" + + fake_runner.available.add("crontab") + fake_runner.handlers["crontab"] = crontab + backend = CronBackend(fake_runner, _make_paths(tmp_path)) + job = backend.discover([Scope.USER])[0] + + plan = backend.plan_update( + job, + JobSpec( + name="backup", + backend=Backend.CRON, + scope=Scope.USER, + command=job.command, + cron_expression="0 2 * * *", + environment=(), + replace_environment=True, + ), + ) + + assert "0 2 * * * /usr/bin/printf '50\\%s\\n'" in plan.payload["new_text"] + assert "OLD=old" not in plan.payload["new_text"] + assert "50\\\\%" not in plan.payload["new_text"] + + def test_plan_update_system_cron_preserves_run_as_and_other_content(tmp_path, fake_runner, monkeypatch) -> None: paths = _make_paths(tmp_path) path = _write( @@ -271,6 +770,42 @@ def test_plan_update_system_cron_preserves_run_as_and_other_content(tmp_path, fa assert trusted_directories == [(paths.cron_d, 0)] +def test_plan_update_system_cron_replaces_environment_without_reescaping_command( + tmp_path, + fake_runner, + monkeypatch, +) -> None: + paths = _make_paths(tmp_path) + _write( + tmp_path / "etc/cron.d/schedls-cleanup", + renderer.render_system_file( + "cleanup", + ["30 3 * * 0 root OLD=old /usr/bin/printf '50\\%s\\n'"], + ), + ) + backend = CronBackend(fake_runner, paths) + job = backend.discover([Scope.SYSTEM])[0] + monkeypatch.setattr(os, "geteuid", lambda: 0) + monkeypatch.setattr("schedls.backends.cron.check_trusted_directory", lambda path, *, expected_uid: None) + + plan = backend.plan_update( + job, + JobSpec( + name="cleanup", + backend=Backend.CRON, + scope=Scope.SYSTEM, + command=job.command, + cron_expression="30 3 * * 0", + environment=(("MESSAGE", "50% off"),), + replace_environment=True, + ), + ) + + assert "30 3 * * 0 root MESSAGE='50\\% off' /usr/bin/printf '50\\%s\\n'" in plan.files[0].content + assert "OLD=old" not in plan.files[0].content + assert "50\\\\%" not in plan.files[0].content + + def test_plan_update_system_cron_requires_root(tmp_path, fake_runner, monkeypatch) -> None: paths = _make_paths(tmp_path) _write( @@ -339,6 +874,32 @@ def test_apply_file_mode_create_and_remove(tmp_path, fake_runner) -> None: assert not path.exists() +def test_apply_file_mode_verifies_disabled_cron_job(tmp_path, fake_runner) -> None: + paths = _make_paths(tmp_path) + path = _write( + tmp_path / "etc/cron.d/schedls-backup", + renderer.render_system_file("backup", ["0 2 * * * root /usr/bin/backup"]), + ) + disabled_content = renderer.render_system_file("backup", ["# schedls:disabled 0 2 * * * root /usr/bin/backup"]) + plan = Plan(backend="cron", action="disable") + plan.files = [FileChange(path=str(path), content=disabled_content, expected_uid=os.getuid())] + plan.payload = { + "mode": "file", + "action": "disable", + "name": "backup", + "path": str(path), + "expected_uid": os.getuid(), + "snapshots": {str(path): path.read_text()}, + "expected_line": "# schedls:disabled 0 2 * * * root /usr/bin/backup", + } + + result = CronBackend(fake_runner, paths).apply(plan) + + assert result.changed is True + assert result.messages == ["System cron job disabled."] + assert path.read_text() == disabled_content + + def test_apply_file_update_refuses_changed_snapshot(tmp_path, fake_runner) -> None: paths = _make_paths(tmp_path) _write(tmp_path / "etc/cron.d/schedls-backup", "changed\n") @@ -423,3 +984,5 @@ def test_plan_remove_refuses_unmanaged(tmp_path, fake_runner) -> None: job = backend.discover([Scope.SYSTEM])[0] with pytest.raises(SafetyRefusalError): backend.plan_remove(job) + with pytest.raises(SafetyRefusalError): + backend.plan_set_enabled(job, False) diff --git a/tests/unit/test_crontab.py b/tests/unit/test_crontab.py index 2c94a38..a18cd3d 100644 --- a/tests/unit/test_crontab.py +++ b/tests/unit/test_crontab.py @@ -82,6 +82,24 @@ def test_with_updated_job_preserves_all_other_content() -> None: ) +def test_toggling_managed_job_preserves_escaped_command_text() -> None: + original = "# schedls:begin name=backup\n0 2 * * * /usr/bin/printf '50\\%s\\n'\n# schedls:end name=backup\n" + + disabled = CrontabDocument(original).with_toggled_enabled("backup", False) + + assert "# schedls:disabled 0 2 * * * /usr/bin/printf '50\\%s\\n'" in disabled + assert CrontabDocument(disabled).with_toggled_enabled("backup", True) == original + + +def test_toggling_managed_job_preserves_absent_trailing_newline() -> None: + original = "# schedls:begin name=backup\n0 2 * * * /usr/bin/backup\n# schedls:end name=backup" + + disabled = CrontabDocument(original).with_toggled_enabled("backup", False) + + assert not disabled.endswith("\n") + assert CrontabDocument(disabled).with_toggled_enabled("backup", True) == original + + def test_with_updated_job_requires_exactly_one_job() -> None: document = CrontabDocument( "# schedls:begin name=backup\n" @@ -111,6 +129,53 @@ def test_malformed_marker_blocks_mutation() -> None: document.with_updated_job("broken", "0 3 * * * /bin/true") +def test_malformed_disabled_marker_blocks_mutation() -> None: + document = CrontabDocument( + "# schedls:begin name=backup\n# schedls:disabled not a cron job\n# schedls:end name=backup\n" + ) + + assert document.has_malformed_markers() + assert "line 2" in document.warnings[0] + with pytest.raises(SafetyRefusalError): + document.with_toggled_enabled("backup", True) + + +def test_indented_disabled_marker_inside_managed_block_blocks_mutation() -> None: + document = CrontabDocument( + "# schedls:begin name=backup\n # schedls:disabled 0 2 * * * /usr/bin/backup\n# schedls:end name=backup\n" + ) + + assert document.has_malformed_markers() + with pytest.raises(SafetyRefusalError): + document.with_toggled_enabled("backup", True) + + +@pytest.mark.parametrize("marker", ["# schedls:disabled", "# schedls:disabled "]) +def test_empty_disabled_marker_inside_managed_block_blocks_mutation(marker: str) -> None: + document = CrontabDocument(f"# schedls:begin name=backup\n{marker}\n# schedls:end name=backup\n") + + assert document.has_malformed_markers() + with pytest.raises(SafetyRefusalError): + document.with_toggled_enabled("backup", True) + + +def test_malformed_disabled_marker_outside_managed_block_does_not_block_mutation() -> None: + document = CrontabDocument("# schedls:disabled not a cron job\n") + + assert not document.has_malformed_markers() + assert "name=backup" in document.with_block("backup", ["0 2 * * * /usr/bin/backup"]) + + +def test_unsafe_environment_prefix_inside_managed_block_blocks_mutation() -> None: + document = CrontabDocument( + "# schedls:begin name=backup\n0 2 * * * VALUE=bad\x00value /usr/bin/backup\n# schedls:end name=backup\n" + ) + + assert document.has_malformed_markers() + with pytest.raises(SafetyRefusalError): + document.with_updated_job("backup", "30 3 * * * /usr/bin/backup") + + def test_empty_crontab() -> None: document = CrontabDocument("") assert document.render() == "" diff --git a/tests/unit/test_operations.py b/tests/unit/test_operations.py index e601b07..e41e9a0 100644 --- a/tests/unit/test_operations.py +++ b/tests/unit/test_operations.py @@ -118,6 +118,30 @@ def test_render_show_explains_unavailable_cron_history() -> None: assert status.index("active") == source.index("current user's crontab") +def test_render_show_displays_cron_environment() -> None: + job = ScheduledJob( + name="cron-1", + backend=Backend.CRON, + scope=Scope.USER, + managed=True, + enabled=True, + schedule=Schedule(ScheduleKind.CRON, "0 2 * * *"), + command=Command(raw="/usr/bin/backup"), + source=JobSource("current user's crontab", line=1), + cron=CronDetails( + expression="0 2 * * *", + line=1, + environment=(("GREETING", "hello world"),), + ), + ) + output, stream = _output() + + inspect_ops.render_show(output, job) + + assert "Environment" in stream.getvalue() + assert "GREETING=hello world" in stream.getvalue() + + def test_render_show_keeps_cron_next_run_null_in_json() -> None: job = ScheduledJob( name="cron-1", @@ -329,7 +353,13 @@ def test_plan_update_refuses_unmanaged() -> None: backend.plan_update(unmanaged, spec) -def _managed_job(name: str = "backup", *, scope: Scope = Scope.USER, service_unit: str | None = None) -> ScheduledJob: +def _managed_job( + name: str = "backup", + *, + scope: Scope = Scope.USER, + service_unit: str | None = None, + environment: tuple[tuple[str, str], ...] = (), +) -> ScheduledJob: return replace( _job(name), scope=scope, @@ -337,17 +367,25 @@ def _managed_job(name: str = "backup", *, scope: Scope = Scope.USER, service_uni timer_unit=f"schedls-{name}.timer", service_unit=service_unit or f"schedls-{name}.service", on_calendar=("*-*-* 02:00:00",), + environment=environment, ), ) -def _spec(name: str = "backup") -> JobSpec: +def _spec( + name: str = "backup", + *, + environment: tuple[tuple[str, str], ...] = (), + replace_environment: bool = False, +) -> JobSpec: return JobSpec( name=name, backend=Backend.SYSTEMD, scope=Scope.USER, command=Command(argv=("/bin/true",)), calendar=("*-*-* 03:00:00",), + environment=environment, + replace_environment=replace_environment, ) @@ -373,6 +411,43 @@ def test_plan_update_derives_unit_names(tmp_path, monkeypatch) -> None: assert "evil" not in " ".join(change.path for change in plan.files) +def test_plan_update_systemd_can_clear_environment(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + backend = SystemdBackend(_analyze_runner()) + + plan = backend.plan_update( + _managed_job(environment=(("OLD", "value"),)), + _spec(environment=(), replace_environment=True), + ) + + service = next(change.content for change in plan.files if change.path.endswith(".service")) + assert "Environment=" not in service + + +def test_plan_update_systemd_replaces_environment(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + backend = SystemdBackend(_analyze_runner()) + + plan = backend.plan_update( + _managed_job(environment=(("OLD", "value"),)), + _spec(environment=(("NEW", "value"),), replace_environment=True), + ) + + service = next(change.content for change in plan.files if change.path.endswith(".service")) + assert 'Environment="NEW=value"' in service + assert "OLD=value" not in service + + +def test_plan_update_systemd_preserves_environment_without_replacement(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + backend = SystemdBackend(_analyze_runner()) + + plan = backend.plan_update(_managed_job(environment=(("OLD", "value"),)), _spec()) + + service = next(change.content for change in plan.files if change.path.endswith(".service")) + assert 'Environment="OLD=value"' in service + + def test_verify_units_rejects_traversal() -> None: backend = SystemdBackend(_analyze_runner()) with pytest.raises(SafetyRefusalError): diff --git a/tests/unit/test_prompt.py b/tests/unit/test_prompt.py index 11c219a..4d10954 100644 --- a/tests/unit/test_prompt.py +++ b/tests/unit/test_prompt.py @@ -47,7 +47,7 @@ def make_prompter(answers: list[str], *, calendar_validator=None) -> tuple[Promp return Prompter(output=output, input_fn=fake, calendar_validator=calendar_validator), fake -def _systemd_job() -> ScheduledJob: +def _systemd_job(*, environment: tuple[tuple[str, str], ...] = ()) -> ScheduledJob: return ScheduledJob( name="backup", backend=Backend.SYSTEMD, @@ -57,11 +57,11 @@ def _systemd_job() -> ScheduledJob: schedule=Schedule(ScheduleKind.CALENDAR, "*-*-* 02:00:00"), command=Command(argv=("/usr/local/bin/backup",)), source=JobSource(detail="systemd user timer"), - systemd=SystemdDetails(on_calendar=("*-*-* 02:00:00",)), + systemd=SystemdDetails(on_calendar=("*-*-* 02:00:00",), environment=environment), ) -def _cron_job() -> ScheduledJob: +def _cron_job(*, environment: tuple[tuple[str, str], ...] = ()) -> ScheduledJob: return ScheduledJob( name="cleanup", backend=Backend.CRON, @@ -71,7 +71,7 @@ def _cron_job() -> ScheduledJob: schedule=Schedule(ScheduleKind.CRON, "0 2 * * *"), command=Command(raw="/usr/local/bin/cleanup"), source=JobSource(detail="current user's crontab"), - cron=CronDetails(expression="0 2 * * *"), + cron=CronDetails(expression="0 2 * * *", environment=environment), ) @@ -165,7 +165,7 @@ def test_eof_raises_confirmation_required() -> None: def test_wizard_new_systemd(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) args = build_parser().parse_args(["new", "-i"]) - answers = ["backup", "", "", "n", "/usr/local/bin/backup /srv/data", "", "02:00", "n"] + answers = ["backup", "", "", "n", "/usr/local/bin/backup /srv/data", "", "02:00", "n", "n"] prompter, _ = make_prompter(answers) args, tail = _wizard_new(args, [], prompter) spec = _spec_for_new(args, tail) @@ -179,7 +179,7 @@ def test_wizard_new_systemd(monkeypatch: pytest.MonkeyPatch) -> None: def test_wizard_new_cron(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) args = build_parser().parse_args(["new", "-i"]) - answers = ["cleanup", "2", "", "n", "/usr/local/bin/cleanup", "", "04:00"] + answers = ["cleanup", "2", "", "n", "/usr/local/bin/cleanup", "", "04:00", "n"] prompter, _ = make_prompter(answers) args, tail = _wizard_new(args, [], prompter) spec = _spec_for_new(args, tail) @@ -189,20 +189,31 @@ def test_wizard_new_cron(monkeypatch: pytest.MonkeyPatch) -> None: assert spec.run_as is None +def test_wizard_new_cron_collects_environment(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) + args = build_parser().parse_args(["new", "cleanup", "-i", "--cron", "--daily", "04:00"]) + prompter, _ = make_prompter(["", "n", "/usr/local/bin/cleanup", "y", "GREETING=hello world", ""]) + + args, tail = _wizard_new(args, [], prompter) + spec = _spec_for_new(args, tail) + + assert spec.environment == (("GREETING", "hello world"),) + + def test_wizard_new_skips_provided_fields(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) args = build_parser().parse_args(["new", "backup", "-i", "--timer", "--daily", "02:00", "--user"]) - prompter, fake = make_prompter(["n"]) + prompter, fake = make_prompter(["n", "n"]) args, tail = _wizard_new(args, ["/bin/true"], prompter) spec = _spec_for_new(args, tail) assert spec.calendar == ("*-*-* 02:00:00",) - assert fake.prompts == ["Set advanced timer options? [y/N] "] + assert fake.prompts == ["Set advanced timer options? [y/N] ", "Add environment variables? [y/N] "] def test_wizard_new_scope_prompt_defaults_to_user(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) args = build_parser().parse_args(["new", "cleanup", "-i", "--cron", "--daily", "04:00"]) - prompter, fake = make_prompter(["", "n", "/usr/local/bin/cleanup"]) + prompter, fake = make_prompter(["", "n", "/usr/local/bin/cleanup", "n"]) args, tail = _wizard_new(args, [], prompter) spec = _spec_for_new(args, tail) assert spec.scope is Scope.USER @@ -214,7 +225,7 @@ def test_wizard_new_scope_prompt_defaults_to_user(monkeypatch: pytest.MonkeyPatc def test_wizard_new_system_cron_prompts_run_as(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) args = build_parser().parse_args(["new", "report", "-i", "--cron", "--daily", "04:00"]) - prompter, _ = make_prompter(["2", "www-data", "n", "/usr/local/bin/report"]) + prompter, _ = make_prompter(["2", "www-data", "n", "/usr/local/bin/report", "n"]) args, tail = _wizard_new(args, [], prompter) spec = _spec_for_new(args, tail) assert spec.scope is Scope.SYSTEM @@ -224,7 +235,7 @@ def test_wizard_new_system_cron_prompts_run_as(monkeypatch: pytest.MonkeyPatch) def test_wizard_new_system_cron_run_as_defaults_to_root(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) args = build_parser().parse_args(["new", "report", "-i", "--cron", "--daily", "04:00"]) - prompter, _ = make_prompter(["2", "", "n", "/usr/local/bin/report"]) + prompter, _ = make_prompter(["2", "", "n", "/usr/local/bin/report", "n"]) args, tail = _wizard_new(args, [], prompter) spec = _spec_for_new(args, tail) assert spec.scope is Scope.SYSTEM @@ -234,7 +245,7 @@ def test_wizard_new_system_cron_run_as_defaults_to_root(monkeypatch: pytest.Monk def test_wizard_new_system_cron_run_as_reprompts(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) args = build_parser().parse_args(["new", "report", "-i", "--cron", "--daily", "04:00"]) - prompter, fake = make_prompter(["2", "bad user", "www-data", "n", "/usr/local/bin/report"]) + prompter, fake = make_prompter(["2", "bad user", "www-data", "n", "/usr/local/bin/report", "n"]) args, tail = _wizard_new(args, [], prompter) spec = _spec_for_new(args, tail) assert spec.run_as == "www-data" @@ -246,7 +257,7 @@ def test_wizard_new_provided_scope_and_run_as_are_not_reasked(monkeypatch: pytes args = build_parser().parse_args( ["new", "report", "-i", "--cron", "--system", "--run-as", "root", "--daily", "04:00"] ) - prompter, fake = make_prompter(["n", "/usr/local/bin/report"]) + prompter, fake = make_prompter(["n", "/usr/local/bin/report", "n"]) args, tail = _wizard_new(args, [], prompter) spec = _spec_for_new(args, tail) assert spec.scope is Scope.SYSTEM @@ -257,7 +268,7 @@ def test_wizard_new_provided_scope_and_run_as_are_not_reasked(monkeypatch: pytes def test_wizard_new_systemd_scope_prompt(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) args = build_parser().parse_args(["new", "backup", "-i", "--timer", "--daily", "02:00"]) - prompter, _ = make_prompter(["2", "n", "/usr/local/bin/backup", "n"]) + prompter, _ = make_prompter(["2", "n", "/usr/local/bin/backup", "n", "n"]) args, tail = _wizard_new(args, [], prompter) spec = _spec_for_new(args, tail) assert spec.scope is Scope.SYSTEM @@ -268,7 +279,7 @@ def test_wizard_edit_changes_schedule(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) job = _systemd_job() args = build_parser().parse_args(["edit", "backup", "-i"]) - answers = ["n", "y", "", "03:00", "n"] + answers = ["n", "y", "", "03:00", "n", "n"] prompter, _ = make_prompter(answers) args, tail = _wizard_edit(job, args, [], prompter) spec = _spec_for_edit(job, args, tail) @@ -278,19 +289,67 @@ def test_wizard_edit_changes_schedule(monkeypatch: pytest.MonkeyPatch) -> None: def test_wizard_edit_changes_cron_schedule(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) args = build_parser().parse_args(["edit", "cleanup", "-i"]) - prompter, _ = make_prompter(["n", "y", "", "03:00"]) + prompter, _ = make_prompter(["n", "y", "", "03:00", "n"]) args, tail = _wizard_edit(_cron_job(), args, [], prompter) spec = _spec_for_edit(_cron_job(), args, tail) assert spec.cron_expression == "0 3 * * *" +def test_wizard_edit_cron_can_clear_environment(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) + job = _cron_job(environment=(("OLD", "value"),)) + args = build_parser().parse_args(["edit", "cleanup", "-i"]) + prompter, _ = make_prompter(["n", "n", "y", ""]) + + args, tail = _wizard_edit(job, args, [], prompter) + spec = _spec_for_edit(job, args, tail) + + assert args.clear_env is True + assert spec.environment == () + + +def test_wizard_edit_cron_replaces_environment(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) + job = _cron_job(environment=(("OLD", "value"),)) + args = build_parser().parse_args(["edit", "cleanup", "-i"]) + prompter, _ = make_prompter(["n", "n", "y", "NEW=value", ""]) + + args, tail = _wizard_edit(job, args, [], prompter) + spec = _spec_for_edit(job, args, tail) + + assert spec.environment == (("NEW", "value"),) + assert spec.replace_environment is True + + +def test_wizard_edit_systemd_can_clear_environment(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) + job = _systemd_job(environment=(("OLD", "value"),)) + args = build_parser().parse_args(["edit", "backup", "-i"]) + prompter, _ = make_prompter(["n", "n", "n", "y", ""]) + + args, tail = _wizard_edit(job, args, [], prompter) + spec = _spec_for_edit(job, args, tail) + + assert args.clear_env is True + assert spec.environment == () + + def test_wizard_edit_keeps_changes_optional(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr("sys.stdin", FakeStdin(isatty=True)) job = _systemd_job() args = build_parser().parse_args(["edit", "backup", "-i"]) - prompter, _ = make_prompter(["n", "n", "n"]) + prompter, _ = make_prompter(["n", "n", "n", "n"]) args, tail = _wizard_edit(job, args, [], prompter) spec = _spec_for_edit(job, args, tail) assert spec.calendar == ("*-*-* 02:00:00",) assert tail == [] + + +def test_spec_for_edit_systemd_clears_environment() -> None: + args = build_parser().parse_args(["edit", "backup", "--clear-env"]) + + spec = _spec_for_edit(_systemd_job(environment=(("OLD", "value"),)), args, []) + + assert spec.environment == () + assert spec.replace_environment is True diff --git a/tests/unit/test_renderers_cron.py b/tests/unit/test_renderers_cron.py index 5612576..0f77ef7 100644 --- a/tests/unit/test_renderers_cron.py +++ b/tests/unit/test_renderers_cron.py @@ -77,6 +77,64 @@ def test_expression_percent_not_double_escaped() -> None: assert line == "0 2 * * * /bin/echo a\\%b" +def test_render_line_with_execution_local_environment() -> None: + line = renderer.render_line( + "0 2 * * *", + Command(argv=("/bin/echo", "a%b")), + environment=(("GREETING", "hello world"), ("RATE", "100%")), + ) + + assert line == "0 2 * * * GREETING='hello world' RATE=100\\% /bin/echo a\\%b" + + +def test_render_shell_line_with_environment_keeps_leading_assignment_semantics() -> None: + line = renderer.render_line( + "0 2 * * *", + Command(shell=True, raw="printf '%s' \"$GREETING\" | cat"), + environment=(("GREETING", "hello world"),), + ) + + assert line == "0 2 * * * GREETING='hello world' printf '\\%s' \"$GREETING\" | cat" + + +@pytest.mark.parametrize( + "value", + [ + "", + "hello world", + "a'b", + r"back\\slash", + "$HOME", + "#hash", + "a=b", + "unicode-é中😀", + "%", + "50% off", + r"a\%b c", + ], +) +def test_execution_local_environment_prefix_round_trips(value: str) -> None: + command = renderer.render_command(Command(argv=("/bin/true",)), environment=(("VALUE", value),)) + + parsed = renderer.parse_environment_prefix(command) + + assert parsed.malformed is False + assert parsed.environment == (("VALUE", value),) + assert parsed.command == "/bin/true" + + +@pytest.mark.parametrize("value", ["bad\nvalue", "bad\rvalue", "bad\x00value"]) +def test_execution_local_environment_rejects_unsafe_control_characters(value: str) -> None: + with pytest.raises(SafetyRefusalError): + renderer.render_command(Command(argv=("/bin/true",)), environment=(("VALUE", value),)) + + +def test_parse_environment_prefix_rejects_unescaped_percent() -> None: + parsed = renderer.parse_environment_prefix("VALUE='50% off' /bin/true") + + assert parsed.malformed is True + + @pytest.mark.parametrize( "bad", ["", "0 2 * *", "0 2 * * * *", "60 2 * * *", "0 25 * * *", "0 2 32 * *", "@bogus", "a b c d e"],