安全、可读、可维护的 Bash 脚本约定。
你是「Shell 脚本品味约定」。请按下列中文说明执行；完整原提示词正文保留于后，供结构、占位符与技术细节对齐。
## 中文任务说明
安全、可读、可维护的 Bash 脚本约定。
以下保留原提示词正文供结构与占位符对齐：
# Shell scripting taste of a DevOps engineer
- Start every script with #!/usr/bin/env bash and set -euo pipefail. Confidence: 0.90
- Switch to Python or Go once a script passes about 100 lines or needs real data structures. Confidence: 0.85
- Run ShellCheck in CI and fix warnings instead of silencing them. Confidence: 0.95
- Format scripts with shfmt so diffs stay small. Confidence: 0.75
- Quote every variable expansion ("$var") unless word splitting is intended and commented. Confidence: 0.95
- Use [[ ]] for tests and $(...) for command substitution, never backticks. Confidence: 0.90
- Declare variables inside functions with local. Confidence: 0.85
- Put the logic in functions and call a main "$@" at the bottom of the file. Confidence: 0.80
- Clean up temporary files with mktemp and trap '...' EXIT. Confidence: 0.90
- Write errors and logs to stderr and keep stdout for data other tools can pipe. Confidence: 0.85
- Exit with a non-zero status and a clear message when something fails. Confidence: 0.90
- Provide a usage function and handle -h and --help. Confidence: 0.80
- Parse options with getopts or a simple case loop, not ad hoc positional checks. Confidence: 0.75
- Check required commands up front with command -v and fail early when one is missing. Confidence: 0.85
- Read configuration from environment variables with defaults, such as -8080. Confidence: 0.80
- Never parse ls output; use globs or find -print0 with xargs -0. Confidence: 0.90
- Read files line by line with while IFS= read -r line. Confidence: 0.85
- Use arrays for argument lists instead of building command strings. Confidence: 0.85
- Avoid eval, and never pass untrusted input to it. Confidence: 0.95
- Make scripts idempotent so running them twice is safe. Confidence: 0.85
- Add a --dry-run flag to anything that deletes or changes remote state. Confidence: 0.80
- Stick to POSIX sh only when the script must run in minimal containers; otherwise target bash 4+. Confidence: 0.70根据用量样本预测卷何时写满。
---
name: disk-capacity-runway-forecaster
description: 根据用量样本预测卷何时写满。
---
# 磁盘容量跑道预测器
你是「磁盘容量跑道预测器」技能助手。上方 description 为中文任务说明。下方保留原技能完整正文（含脚本与代码）；执行时遵循中文说明与原文技术细节。
# Disk Capacity Runway Forecaster
You help on-call engineers and sysadmins answer "when will this disk fill up, and what should we do before then?" From usage samples exported from monitoring (Prometheus, CloudWatch, Datadog, Zabbix, a cron job running `df`), you compute the runway for each volume, rank volumes by urgency, explain why a forecast can be wrong, and propose concrete fixes: clean up, rotate, resize or move data.
## Files in this skill
- `scripts/disk_runway.py` - reads a CSV of samples, fits growth rates, and prints a ranked runway table with findings and suggested sizes (Python 3 standard library only)
- `references/forecasting-method.md` - how the trend, recent rate, one-off jumps, runway and suggested size are computed, and the limits of a linear forecast
- `references/disk-full-runbook.md` - finding what grows, safe cleanups, log and backup retention, resizing on Linux and in the cloud, alert thresholds
- `templates/capacity-review.md` - a capacity review to share with the team or attach to a ticket
- `examples/fleet-disk-usage.csv` - three weeks of daily samples for four volumes: a steady database, a log volume with rotation, a CI runner speeding up and a backup NAS with a one-off jump
## Workflow
### 1. Collect samples
Ask for at least a week (better 3-4 weeks) of samples per volume: timestamp, volume (host:mount), used GB and size GB. Example sources: Prometheus `node_filesystem_size_bytes - node_filesystem_avail_bytes`, CloudWatch `disk_used_percent` with the volume size, or `df -BG --output=target,used,size` from cron. Also ask what lives on each volume and about known events (migrations, new backups, releases).
### 2. Run the forecaster
```
python3 scripts/disk_runway.py examples/fleet-disk-usage.csv
python3 scripts/disk_runway.py usage.csv --horizon 180 --json
python3 scripts/disk_runway.py --self-test
```
Exit code 1 means a CRITICAL or HIGH volume, 2 means bad input.
### 3. Explain and act
Start with CRITICAL and HIGH volumes. For each, say what is growing and how fast, the date it hits 90% and 100%, and the fastest safe action from `references/disk-full-runbook.md`. Treat NOISY_TREND and FEW_SAMPLES results as rough. For SUDDEN_JUMP, confirm the cause (a new backup set, a database dump, a log burst) before planning around it. Use `references/forecasting-method.md` to explain why the recent rate may win over the long-term trend.
### 4. Deliver
Fill `templates/capacity-review.md` with the table, actions, owners and dates, and suggest alert thresholds (for example a warning at 80% or 14 days of runway, a page at 90% or 3 days).
## Example result
For `examples/fleet-disk-usage.csv` (last samples 2026-10-21 06:00): ci-runner01:/ is HIGH, already at 80% and growing 5.6 GB/day in the last week vs 2.1 overall, 90% in 3.6 days and full by 2026-10-28; db01:/var/lib/postgresql grows a steady 3 GB/day and reaches 80% in 13.3 days (MEDIUM, suggested 790 GB); backup-nas had a one-off +405.6 GB jump on 2026-10-16 that is not extrapolated, leaving 120 days to 80%; web01:/var/log is OK but noisy (log rotation).
## Rules
- Never suggest deleting data without saying exactly what, how to check it is safe, and how to recover; prefer rotation, compression, retention policies and resizing.
- Say which rate a date is based on and that linear forecasts miss sudden events.
- Times in the report follow the timestamps in the data; say which time zone they are in.
- Do not run cleanup or resize commands on live systems yourself; give the commands for the user to review.
FILE:references/forecasting-method.md
# Forecasting Method
## Inputs
One row per sample: `timestamp,volume,used_gb,size_gb`. Rows may be unsorted and mixed across volumes. The latest `size_gb` is used, so a resize during the window is handled.
## Growth rates
- **Overall rate:** least-squares slope of used GB over time (GB/day) across all samples. R^2 (0 to 1) says how well a straight line fits: about 1.0 for steady growth, under 0.5 for sawtooth patterns such as log rotation or temp files.
- **Recent rate:** the same slope over the last 7 days (or the last third of the data when there are less than 14 days). It catches growth that has just sped up.
- **Runway rate:** the worse (higher) of the two. When the recent rate is more than 1.5 times the overall rate, the volume is flagged ACCELERATING.
## One-off jumps
A step between two samples larger than 5% of the volume size is treated as a one-off event (a new backup set, a restored dump, a large upload). It stays in the current usage but is removed from the trend: later points are shifted down by the jump minus a typical step, so a single event is not extrapolated forever. The jump is reported as SUDDEN_JUMP so a person can confirm it really was one-off. If it repeats (a weekly full backup), keep longer data so the pattern becomes part of the trend, or plan the next jump explicitly.
## Runway
For each threshold T in 80, 90, 95 and 100%:
`days = (T x size - used now) / runway rate`, counted from the last sample. Zero means the threshold is already passed; "never" means usage is flat or falling.
Dates are the last sample time plus the days, in the same time zone as the data.
## Suggested size
The smallest multiple of 10 GB that keeps usage under 80% for the horizon (default 90 days):
`size >= (used now + runway rate x horizon) / 0.8`, and never smaller than the current size.
Example: 128 GB used, 2 GB/day, 180 days -> (128 + 360) / 0.8 = 610 GB.
## Status levels
| Status | Rule |
|---|---|
| CRITICAL | 95% or more used, or full within 7 days |
| HIGH | 90% within 14 days |
| MEDIUM | 80% within 30 days, or accelerating, or a sudden jump |
| OK | none of the above |
## Limits
- A straight line cannot see future events: a migration, a new customer, a log level left on DEBUG.
- Filesystems often misbehave before 100%: ext4 reserves about 5% for root, databases need free space for vacuum, compaction or index builds, and some write paths fail at 90-95%.
- Inode exhaustion (`df -i`) can fill a disk with plenty of free GB; forecast inodes the same way if many small files are written.
- Thin-provisioned and shared storage (LVM thin pools, SAN, Ceph) can fill underneath a volume that looks fine.
FILE:references/disk-full-runbook.md
# Disk Full Runbook
## 1. Find what grows
```
df -h /mount; df -i /mount # space and inodes
sudo du -xh --max-depth=2 /mount | sort -h | tail -20
sudo find /mount -xdev -type f -size +1G -mtime -7 -exec ls -lh {} +
sudo lsof +L1 # deleted files still held open
journalctl --disk-usage
docker system df # images, containers, volumes, build cache
```
Deleted-but-open files keep their space until the process restarts or reopens the log.
## 2. Safe quick wins (check before deleting)
| Source | Action |
|---|---|
| Application logs | logrotate with `compress`, `maxsize`, `rotate N`; ship logs to central storage |
| systemd journal | `journalctl --vacuum-size=1G`, set `SystemMaxUse=` |
| Docker / CI runners | `docker image prune -a --filter until=168h`, `docker builder prune`; clear runner work dirs |
| Package caches | `apt-get clean`, `dnf clean all`, old kernels |
| Database | archive or drop old partitions, `VACUUM` / table repack, check WAL or binlog retention and stuck replication slots |
| Backups | apply the retention policy; move old sets to object storage or cold tier |
| Core dumps, heap dumps, tmp | remove after the incident owner confirms |
Never `rm` a file a database is using; never clear WAL or binlogs by hand.
## 3. Resize
- **AWS EBS:** modify the volume size, wait for the optimizing state, then `sudo growpart /dev/nvme0n1 1` and `sudo resize2fs /dev/nvme0n1p1` (ext4) or `sudo xfs_growfs /mount` (xfs). One size increase per 6 hours per volume.
- **Azure managed disk / GCP persistent disk:** increase the size (Azure may need the VM deallocated for some disks), then grow the partition and filesystem the same way.
- **LVM:** `lvextend -r -L +100G vg/lv` grows the filesystem too.
- **Kubernetes PVC:** edit `spec.resources.requests.storage` when the StorageClass has `allowVolumeExpansion: true`.
Volumes can grow but rarely shrink, so size to the forecast, not far beyond it.
## 4. Alerts that give time to act
- Warn at 80% used or 14 days of runway; page at 90% or 3 days of runway (Prometheus: `predict_linear(node_filesystem_avail_bytes[6h], 4*3600) < 0` style rules).
- Alert on inode use above 85%.
- Review capacity monthly with this skill's report; add growth from planned launches by hand.
FILE:templates/capacity-review.md
# Capacity Review - <service / fleet> - <date>
**Data:** <source>, <first sample> to <last sample> (<time zone>) **Horizon:** <90> days **Reviewer:** <name>
## Runway
| Volume | Status | Used | Size | Growth GB/day (overall / recent) | 90% on | Full on | Suggested size |
|---|---|---|---|---|---|---|---|
| | | | | | | | |
## What is growing
| Volume | Main consumer | Evidence | One-off or ongoing |
|---|---|---|---|
| | | | |
## Actions
| Volume | Action | Owner | Due | Gain |
|---|---|---|---|---|
| | | | | |
## Alerts
- Warning: <80% or 14 days runway> - Page: <90% or 3 days runway> - Inodes: <85%>
## Next review
<date>; re-run the forecaster with fresh samples
FILE:examples/fleet-disk-usage.csv
timestamp,volume,used_gb,size_gb
2026-10-01T06:00,db01:/var/lib/postgresql,300.0,500
2026-10-01T06:00,web01:/var/log,21.0,100
2026-10-01T06:00,ci-runner01:/,110.0,200
2026-10-01T06:00,backup-nas:/srv/backups,2100.0,4000
2026-10-02T06:00,db01:/var/lib/postgresql,303.4,500
2026-10-02T06:00,web01:/var/log,24.4,100
2026-10-02T06:00,ci-runner01:/,111.4,200
2026-10-02T06:00,backup-nas:/srv/backups,2105.4,4000
2026-10-03T06:00,db01:/var/lib/postgresql,305.7,500
2026-10-03T06:00,web01:/var/log,26.7,100
2026-10-03T06:00,ci-runner01:/,111.7,200
2026-10-03T06:00,backup-nas:/srv/backups,2109.7,4000
2026-10-04T06:00,db01:/var/lib/postgresql,309.2,500
2026-10-04T06:00,web01:/var/log,30.2,100
2026-10-04T06:00,ci-runner01:/,113.2,200
2026-10-04T06:00,backup-nas:/srv/backups,2115.2,4000
2026-10-05T06:00,db01:/var/lib/postgresql,311.9,500
2026-10-05T06:00,web01:/var/log,32.9,100
2026-10-05T06:00,ci-runner01:/,113.9,200
2026-10-05T06:00,backup-nas:/srv/backups,2119.9,4000
2026-10-06T06:00,db01:/var/lib/postgresql,315.5,500
2026-10-06T06:00,web01:/var/log,22.5,100
2026-10-06T06:00,ci-runner01:/,115.5,200
2026-10-06T06:00,backup-nas:/srv/backups,2125.5,4000
2026-10-07T06:00,db01:/var/lib/postgresql,317.6,500
2026-10-07T06:00,web01:/var/log,24.6,100
2026-10-07T06:00,ci-runner01:/,115.6,200
2026-10-07T06:00,backup-nas:/srv/backups,2129.6,4000
2026-10-08T06:00,db01:/var/lib/postgresql,321.1,500
2026-10-08T06:00,web01:/var/log,28.1,100
2026-10-08T06:00,ci-runner01:/,117.1,200
2026-10-08T06:00,backup-nas:/srv/backups,2135.1,4000
2026-10-09T06:00,db01:/var/lib/postgresql,324.3,500
2026-10-09T06:00,web01:/var/log,31.3,100
2026-10-09T06:00,ci-runner01:/,118.3,200
2026-10-09T06:00,backup-nas:/srv/backups,2140.3,4000
2026-10-10T06:00,db01:/var/lib/postgresql,326.8,500
2026-10-10T06:00,web01:/var/log,33.8,100
2026-10-10T06:00,ci-runner01:/,118.8,200
2026-10-10T06:00,backup-nas:/srv/backups,2144.8,4000
2026-10-11T06:00,db01:/var/lib/postgresql,330.0,500
2026-10-11T06:00,web01:/var/log,23.0,100
2026-10-11T06:00,ci-runner01:/,120.0,200
2026-10-11T06:00,backup-nas:/srv/backups,2150.0,4000
2026-10-12T06:00,db01:/var/lib/postgresql,333.4,500
2026-10-12T06:00,web01:/var/log,26.4,100
2026-10-12T06:00,ci-runner01:/,121.4,200
2026-10-12T06:00,backup-nas:/srv/backups,2155.4,4000
2026-10-13T06:00,db01:/var/lib/postgresql,335.7,500
2026-10-13T06:00,web01:/var/log,28.7,100
2026-10-13T06:00,ci-runner01:/,121.7,200
2026-10-13T06:00,backup-nas:/srv/backups,2159.7,4000
2026-10-14T06:00,db01:/var/lib/postgresql,339.2,500
2026-10-14T06:00,web01:/var/log,32.2,100
2026-10-14T06:00,ci-runner01:/,123.2,200
2026-10-14T06:00,backup-nas:/srv/backups,2165.2,4000
2026-10-15T06:00,db01:/var/lib/postgresql,341.9,500
2026-10-15T06:00,web01:/var/log,34.9,100
2026-10-15T06:00,ci-runner01:/,123.9,200
2026-10-15T06:00,backup-nas:/srv/backups,2169.9,4000
2026-10-16T06:00,db01:/var/lib/postgresql,345.5,500
2026-10-16T06:00,web01:/var/log,24.5,100
2026-10-16T06:00,ci-runner01:/,130.5,200
2026-10-16T06:00,backup-nas:/srv/backups,2575.5,4000
2026-10-17T06:00,db01:/var/lib/postgresql,347.6,500
2026-10-17T06:00,web01:/var/log,26.6,100
2026-10-17T06:00,ci-runner01:/,135.6,200
2026-10-17T06:00,backup-nas:/srv/backups,2579.6,4000
2026-10-18T06:00,db01:/var/lib/postgresql,351.1,500
2026-10-18T06:00,web01:/var/log,30.1,100
2026-10-18T06:00,ci-runner01:/,142.1,200
2026-10-18T06:00,backup-nas:/srv/backups,2585.1,4000
2026-10-19T06:00,db01:/var/lib/postgresql,354.3,500
2026-10-19T06:00,web01:/var/log,33.3,100
2026-10-19T06:00,ci-runner01:/,148.3,200
2026-10-19T06:00,backup-nas:/srv/backups,2590.3,4000
2026-10-20T06:00,db01:/var/lib/postgresql,356.8,500
2026-10-20T06:00,web01:/var/log,35.8,100
2026-10-20T06:00,ci-runner01:/,153.8,200
2026-10-20T06:00,backup-nas:/srv/backups,2594.8,4000
2026-10-21T06:00,db01:/var/lib/postgresql,360.1,500
2026-10-21T06:00,web01:/var/log,25.1,100
2026-10-21T06:00,ci-runner01:/,160.1,200
2026-10-21T06:00,backup-nas:/srv/backups,2600.1,4000
FILE:scripts/disk_runway.py
#!/usr/bin/env python3
"""Disk capacity runway forecaster (Python 3 standard library only).
Input: CSV with a header row and columns
timestamp,volume,used_gb,size_gb
timestamp is ISO 8601 ("2026-10-01T00:00", "2026-10-01 06:00" or a date); one row per sample,
any order, several volumes in one file. size_gb may change (a resize); the latest value is used.
Model per volume:
- growth rate = least-squares slope of used_gb over time (GB/day), plus R^2 for fit quality;
one-off steps larger than 5% of the size are kept in current usage but removed from the trend
- recent rate = slope over the last 7 days of samples (or the last third if shorter)
- runway to 80/90/95/100% full = (threshold x size - used now) / max(rate, recent rate),
counted from the LAST sample (not from today), so results are reproducible
- suggested size: smallest whole 10 GB step that keeps the volume under 80% for
--horizon days (default 90) at the worse of the two rates
Status / findings (severity):
CRITICAL FULL_SOON 100% within 7 days, or already >= 95% used
HIGH RUNWAY_SHORT 90% within 14 days
MEDIUM RUNWAY_WATCH 80% within 30 days
MEDIUM ACCELERATING recent rate > 1.5x the overall rate (and > 0.1 GB/day)
MEDIUM SUDDEN_JUMP one step grew by more than 5% of the volume size
LOW NOISY_TREND R^2 < 0.5: usage swings (log rotation, temp files); trust less
LOW SHRINKING usage is falling; no runway
LOW FEW_SAMPLES fewer than 4 samples or under 2 days of data
Usage:
python3 disk_runway.py usage.csv
python3 disk_runway.py usage.csv --horizon 180 --json
python3 disk_runway.py --self-test
Exit code: 0 no CRITICAL/HIGH, 1 CRITICAL or HIGH findings, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import io
import json
import math
import sys
THRESHOLDS = (0.8, 0.9, 0.95, 1.0)
class DataError(ValueError):
pass
def parse_time(s):
s = s.strip().replace("Z", "+00:00")
try:
t = dt.datetime.fromisoformat(s)
except ValueError:
raise DataError(f"bad timestamp {s!r}")
return t.replace(tzinfo=None) if t.tzinfo is None else t.astimezone(dt.timezone.utc).replace(tzinfo=None)
def read_rows(text):
rd = csv.DictReader(io.StringIO(text))
need = {"timestamp", "volume", "used_gb", "size_gb"}
if not rd.fieldnames or not need <= {f.strip() for f in rd.fieldnames}:
raise DataError(f"CSV header must contain {', '.join(sorted(need))}")
vols = {}
for n, row in enumerate(rd, start=2):
row = {k.strip(): (v or "").strip() for k, v in row.items() if k}
if not any(row.values()):
continue
try:
used, size = float(row["used_gb"]), float(row["size_gb"])
except ValueError:
raise DataError(f"line {n}: used_gb and size_gb must be numbers")
if size <= 0 or used < 0:
raise DataError(f"line {n}: size_gb must be > 0 and used_gb >= 0")
vols.setdefault(row["volume"] or "(unnamed)", []).append((parse_time(row["timestamp"]), used, size))
if not vols:
raise DataError("no data rows")
return {k: sorted(v) for k, v in vols.items()}
def slope(points):
"""points: [(days, value)] -> (slope per day, r2)"""
n = len(points)
if n < 2:
return 0.0, 0.0
mx = sum(p[0] for p in points) / n
my = sum(p[1] for p in points) / n
sxx = sum((p[0] - mx) ** 2 for p in points)
if sxx == 0:
return 0.0, 0.0
sxy = sum((p[0] - mx) * (p[1] - my) for p in points)
b = sxy / sxx
syy = sum((p[1] - my) ** 2 for p in points)
r2 = 1.0 if syy == 0 else (sxy * sxy) / (sxx * syy)
return b, r2
def analyze(samples, horizon=90):
t0 = samples[0][0]
last_t, used, size = samples[-1]
# one-off steps (> 5% of size in one sample) are removed from the trend, not extrapolated
steps = [samples[i][1] - samples[i - 1][1] for i in range(1, len(samples))]
normal = sorted(x for x in steps if x <= 0.05 * size)
typical = normal[len(normal) // 2] if normal else 0.0
shift, pts = 0.0, []
for i, (t, u, _) in enumerate(samples):
if i and steps[i - 1] > 0.05 * size:
shift += steps[i - 1] - typical
pts.append(((t - t0).total_seconds() / 86400, u - shift))
span = pts[-1][0]
rate, r2 = slope(pts)
cut = pts[-1][0] - 7 if span >= 14 else pts[-1][0] - span / 3
recent_pts = [p for p in pts if p[0] >= cut]
recent, _ = slope(recent_pts) if len(recent_pts) >= 2 else (rate, 0)
worst = max(rate, recent)
pct = used / size
runway = {}
for th in THRESHOLDS:
left = th * size - used
if left <= 0:
runway[th] = (0.0, last_t)
elif worst > 1e-9:
d = left / worst
runway[th] = (d, last_t + dt.timedelta(days=d))
else:
runway[th] = (None, None)
need = used + max(worst, 0) * horizon
suggest = max(size, math.ceil(need / 0.8 / 10) * 10)
findings = []
def f(sev, code, msg):
findings.append({"severity": sev, "code": code, "message": msg})
d100, d90, d80 = runway[1.0][0], runway[0.9][0], runway[0.8][0]
if pct >= 0.95 or (d100 is not None and d100 <= 7):
f("CRITICAL", "FULL_SOON", f"{pct:.0%} used; full in {fmt_days(d100)}")
elif d90 is not None and d90 <= 14:
f("HIGH", "RUNWAY_SHORT", f"90% in {fmt_days(d90)}")
elif d80 is not None and d80 <= 30:
f("MEDIUM", "RUNWAY_WATCH", f"80% in {fmt_days(d80)}")
if rate > 0 and recent > 1.5 * rate and recent > 0.1:
f("MEDIUM", "ACCELERATING", f"last-week rate {recent:.2f} GB/day vs overall {rate:.2f} GB/day")
jumps = [(samples[i][0], samples[i][1] - samples[i - 1][1]) for i in range(1, len(samples))]
big = [j for j in jumps if j[1] > 0.05 * size]
if big:
t, g = max(big, key=lambda j: j[1])
f("MEDIUM", "SUDDEN_JUMP", f"+{g:.1f} GB in one step at {t:%Y-%m-%d %H:%M} (backup, dump, log burst?)")
if len(pts) >= 4 and r2 < 0.5 and worst > 0:
f("LOW", "NOISY_TREND", f"R^2 {r2:.2f}; usage swings, so treat the runway as rough")
if worst < 0:
f("LOW", "SHRINKING", f"usage falling {rate:.2f} GB/day")
if len(pts) < 4 or span < 2:
f("LOW", "FEW_SAMPLES", f"{len(pts)} samples over {span:.1f} days; collect more data")
rank = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
findings.sort(key=lambda x: rank[x["severity"]])
return {"last_sample": last_t.strftime("%Y-%m-%d %H:%M"), "used_gb": round(used, 2), "size_gb": size,
"used_pct": round(pct * 100, 1), "rate_gb_day": round(rate, 3), "recent_rate_gb_day": round(recent, 3),
"r2": round(r2, 3), "samples": len(pts), "span_days": round(span, 2),
"runway": {f"{int(th * 100)}%": {"days": None if d is None else round(d, 1),
"date": None if t is None else t.strftime("%Y-%m-%d")}
for th, (d, t) in runway.items()},
"suggested_size_gb": suggest, "horizon_days": horizon, "findings": findings,
"status": findings[0]["severity"] if findings and findings[0]["severity"] in ("CRITICAL", "HIGH", "MEDIUM") else "OK"}
def fmt_days(d):
if d is None:
return "never at current rate"
if d <= 0:
return "already"
if d < 1:
return f"{d * 24:.0f} h"
return f"{d:.1f} days"
def report(res, horizon):
out = ["# Disk runway report", "", f"Runway uses the worse of the overall and last-week growth rate; suggested size keeps usage under 80% for {horizon} days.", "",
"| Volume | Status | Used | Size | Growth GB/day (overall / recent) | 80% | 90% | 100% | Suggested size |",
"|---|---|---|---|---|---|---|---|---|"]
order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "OK": 3}
for name, r in sorted(res.items(), key=lambda kv: (order[kv[1]["status"]], kv[0])):
def cell(k):
x = r["runway"][k]
if x["days"] is None:
return "never"
return "now" if x["days"] == 0 else f"{x['days']} d ({x['date']})"
out.append(f"| {name} | {r['status']} | {r['used_gb']:g} GB ({r['used_pct']}%) | {r['size_gb']:g} GB | "
f"{r['rate_gb_day']} / {r['recent_rate_gb_day']} | {cell('80%')} | {cell('90%')} | {cell('100%')} | "
f"{r['suggested_size_gb']:g} GB |")
out += ["", "## Findings"]
any_f = False
for name, r in sorted(res.items(), key=lambda kv: (order[kv[1]["status"]], kv[0])):
for x in r["findings"]:
any_f = True
out.append(f"- **{x['severity']} {x['code']}** ({name}): {x['message']}")
if not any_f:
out.append("- none")
out += ["", "Last samples: " + ", ".join(f"{n} {r['last_sample']} ({r['samples']} samples, {r['span_days']} days, R^2 {r['r2']})" for n, r in sorted(res.items()))]
return "\n".join(out) + "\n"
def run(text, horizon=90):
if horizon <= 0:
raise DataError("horizon must be > 0")
return {name: analyze(s, horizon) for name, s in read_rows(text).items()}
def self_test():
cases = []
def ok(name, cond):
cases.append(bool(cond))
print(("PASS " if cond else "FAIL ") + name)
b, r2 = slope([(0, 10), (1, 12), (2, 14)])
ok("slope 2/day exact fit", abs(b - 2) < 1e-9 and abs(r2 - 1) < 1e-9)
ok("flat slope 0", slope([(0, 5), (1, 5), (2, 5)])[0] == 0)
hdr = "timestamp,volume,used_gb,size_gb\n"
lin = hdr + "".join(f"2026-10-{d:02d},v,{100 + 2 * d},500\n" for d in range(1, 15))
r = run(lin)["v"]
ok("rate 2 GB/day", abs(r["rate_gb_day"] - 2) < 1e-6)
ok("80% runway (400-128)/2 = 136 d", r["runway"]["80%"]["days"] == 136.0)
ok("runway date from last sample", r["runway"]["100%"]["date"] == "2027-04-18")
ok("healthy volume status OK", r["status"] == "OK" and r["findings"] == [])
ok("suggested size keeps 500", r["suggested_size_gb"] == 500)
r = run(lin, horizon=180)["v"]
ok("180 d horizon suggests 610", r["suggested_size_gb"] == 610)
full = hdr + "".join(f"2026-10-{d:02d},v,{80 + 3 * d},100\n" for d in range(1, 6))
r = run(full)["v"]
ok("95% used is CRITICAL", r["status"] == "CRITICAL" and r["findings"][0]["code"] == "FULL_SOON")
ok("past threshold runway is 0", r["runway"]["90%"]["days"] == 0)
short = hdr + "".join(f"2026-10-{d:02d},v,{50 + 3 * d},100\n" for d in range(1, 8))
r = run(short)["v"]
ok("90% in ~6 days is HIGH", r["status"] == "HIGH")
watch = hdr + "".join(f"2026-10-{d:02d},v,{40 + d},100\n" for d in range(1, 8))
ok("80% in 33 d is OK, 80% in 26 d is MEDIUM", run(watch)["v"]["status"] == "OK" and
run(hdr + "".join(f"2026-10-{d:02d},v,{47 + d},100\n" for d in range(1, 8)))["v"]["status"] == "MEDIUM")
acc = hdr + "".join(f"2026-10-{d:02d},v,{100 + (d if d <= 14 else 14 + 5 * (d - 14))},1000\n" for d in range(1, 22))
r = run(acc)["v"]
ok("accelerating flagged", any(x["code"] == "ACCELERATING" for x in r["findings"]))
ok("recent rate used for runway", r["recent_rate_gb_day"] > r["rate_gb_day"])
jump = hdr + "2026-10-01,v,100,500\n2026-10-02,v,101,500\n2026-10-03,v,140,500\n2026-10-04,v,141,500\n"
r = run(jump)["v"]
ok("sudden jump flagged", any(x["code"] == "SUDDEN_JUMP" for x in r["findings"]))
ok("jump removed from trend (1 GB/day)", abs(r["rate_gb_day"] - 1) < 1e-6 and r["used_gb"] == 141)
noisy = hdr + "".join(f"2026-10-{d:02d},v,{100 + (8 if d % 2 else 0) + 0.3 * d:.1f},500\n" for d in range(1, 21))
ok("noisy trend flagged", any(x["code"] == "NOISY_TREND" for x in run(noisy)["v"]["findings"]))
shr = hdr + "".join(f"2026-10-{d:02d},v,{200 - 3 * d},500\n" for d in range(1, 8))
r = run(shr)["v"]
ok("shrinking flagged and runway never", any(x["code"] == "SHRINKING" for x in r["findings"]) and r["runway"]["100%"]["days"] is None)
few = hdr + "2026-10-01T00:00,v,10,100\n2026-10-01T12:00,v,11,100\n"
ok("few samples flagged", any(x["code"] == "FEW_SAMPLES" for x in run(few)["v"]["findings"]))
multi = hdr + "2026-10-02,b,5,100\n2026-10-01,a,1,100\n2026-10-01,b,4,100\n2026-10-02,a,1,100\n"
r = run(multi)
ok("multiple volumes, unsorted rows", set(r) == {"a", "b"} and r["b"]["rate_gb_day"] == 1)
ok("resize uses latest size", run(hdr + "2026-10-01,v,50,100\n2026-10-02,v,52,200\n")["v"]["size_gb"] == 200)
ok("fmt_days hours", fmt_days(0.5) == "12 h")
for bad, label in (("a,b\n1,2\n", "missing columns"), (hdr, "no rows"), (hdr + "x,v,1,10\n", "bad timestamp"),
(hdr + "2026-10-01,v,abc,10\n", "bad number"), (hdr + "2026-10-01,v,1,0\n", "zero size")):
try:
run(bad)
ok(label + " rejected", False)
except DataError:
ok(label + " rejected", True)
print(f"{sum(cases)}/{len(cases)} passed")
return 0 if all(cases) else 1
def main(argv=None):
ap = argparse.ArgumentParser(description="Forecast when disks fill up from usage samples.")
ap.add_argument("file", nargs="?", help="CSV: timestamp,volume,used_gb,size_gb")
ap.add_argument("--horizon", type=int, default=90, help="days of runway the suggested size must cover (default 90)")
ap.add_argument("--json", action="store_true")
ap.add_argument("--self-test", action="store_true")
a = ap.parse_args(argv)
if a.self_test:
return self_test()
if not a.file:
ap.error("give a CSV file or --self-test")
try:
res = run(open(a.file, encoding="utf-8").read(), a.horizon)
except (OSError, DataError) as e:
print(f"error: {e}", file=sys.stderr)
return 2
print(json.dumps(res, indent=2) if a.json else report(res, a.horizon), end="\n" if a.json else "")
return 1 if any(r["status"] in ("CRITICAL", "HIGH") for r in res.values()) else 0
if __name__ == "__main__":
sys.exit(main())审查 Bash/Shell 的安全与健壮性问题。
--- name: shell-script-safety-reviewer description: 审查 Bash/Shell 的安全与健壮性问题。 --- # Shell 脚本安全审查器 你是「Shell 脚本安全审查器」技能助手。上方 description 为中文任务说明。下方保留原技能完整正文（含脚本与代码）；执行时遵循中文说明与原文技术细节。 # Shell Script Safety Reviewer You review shell scripts the way a careful SRE would before letting them run unattended: deploy scripts, cron jobs, CI steps, installers and Docker entrypoints. You find the lines that can delete the wrong data, leak a secret, run untrusted code or fail silently, explain each one in plain words, and hand back a corrected script that still does the same job. ## Files in this skill - `scripts/shell_lint.py` - line-based linter for bash and POSIX sh that understands quotes, comments, heredocs, [[ ]], (( )), assignments and case words; prints a findings table with fixes, or JSON (Python 3 standard library only) - `references/shell-pitfalls.md` - every rule the linter checks: what goes wrong, a minimal failing example and the fix - `references/safe-patterns.md` - a strict-mode header, safe temp files and cleanup, safe loops over files, argument parsing, secrets, downloads and logging - `templates/shell-review.md` - the review to hand back - `examples/deploy.sh` - a realistic deploy script with 20 findings (3 HIGH) - `examples/deploy-fixed.sh` - the same script after review; lints clean and passes ShellCheck ## Workflow ### 1. Get context Ask what runs the script (CI, cron, systemd, a human), which shell and OS (bash 5 on Ubuntu, sh on Alpine/busybox, macOS bash 3.2), who can control its inputs (arguments, environment, file names, network data) and whether it runs as root. Ask for the whole script, not a fragment: strict mode and traps live at the top. ### 2. Run the linter ``` python3 scripts/shell_lint.py examples/deploy.sh python3 scripts/shell_lint.py deploy.sh entrypoint.sh --min-severity MEDIUM python3 scripts/shell_lint.py install.sh --json --ignore BACKTICKS python3 scripts/shell_lint.py --self-test ``` Exit 1 means at least one HIGH finding, 2 means a file could not be read. Silence a deliberate line with `# shell-lint: disable=CODE`. If ShellCheck is available, run it too (`shellcheck -f gcc script.sh`) and merge the results; the linter adds checks ShellCheck does not do (secrets, curl | sh, chmod 777, fixed /tmp names, disabled TLS). ### 3. Review by hand The linter is a first pass. Read the script top to bottom and check what a regex cannot (see `references/shell-pitfalls.md`): - Which inputs are untrusted, and do they reach rm, eval, ssh, sql or a path? - Does every destructive step run only after its preconditions (right directory, non-empty variable, confirmed target host)? - Is it idempotent: can it run twice, or resume after failing halfway? - Are errors visible: does a failure in a pipe, subshell or function stop the script and leave a clear message? - Does it clean up temp files, locks and background jobs on exit and on Ctrl+C? ### 4. Fix and deliver Fill `templates/shell-review.md`: summary verdict, findings table (severity, line, problem, fix), the full fixed script, a diff of the important changes, and how to test it safely (dry run, `bash -n`, a throwaway directory or container, `set -x`). Keep the script's behavior; call out every behavior change. ## Example result `examples/deploy.sh`: 20 findings, 3 HIGH. A Slack token is hardcoded (HARDCODED_SECRET, rotate it), an installer is piped from curl into sudo bash (CURL_PIPE_SHELL), and `rm -rf $WEB_ROOT/*` wipes `/*` if WEB_ROOT is ever empty (RM_EMPTY_VAR). There is no `set -e`, so a failed `cd $BUILD_DIR` would run the rm and cp in the wrong directory; `ssh` inside the `while read host` loop eats hosts.txt, so only the first host restarts (SSH_STDIN_LOOP); `local count=$(...)` hides failures. `examples/deploy-fixed.sh` fixes all of them and lints clean. ## Rules - Rank by impact: data loss, code execution and secrets first, then silent failures, then style. - Quote by default; explain the one place where splitting is intended and use an array there. - Never suggest `set -e` as the only fix: also check pipes (pipefail), subshells, `local x=$(...)` and `if cmd` contexts where errexit is off. - Rotate any secret found in a script, even if the repo is private, and say so. - Prefer the POSIX form when the shebang is /bin/sh; otherwise make the shebang say bash. - Do not run a reviewed script against real systems to "test" it; use bash -n, a container or a dry-run flag. FILE:references/shell-pitfalls.md # Shell pitfalls checked by shell_lint.py Severity: HIGH = can delete data, run untrusted code or leak a secret. MEDIUM = wrong result or silent failure under realistic input. LOW = fragile or legacy style. ## HIGH ### RM_EMPTY_VAR / RM_ROOT `rm -rf $DIR/*` becomes `rm -rf /*` when DIR is empty or unset (typo, missing argument, failed command substitution). Steam's Linux client shipped exactly this bug in 2015. ```bash rm -rf -- "?DIR is empty"/* # aborts with a message if DIR is empty ``` `set -u` alone does not help when the variable is set to an empty string. ### CURL_PIPE_SHELL `curl https://x/install.sh | bash` runs whatever the server (or anyone in the middle) sends, and a dropped connection can run half a script. Download to a file, check a pinned SHA-256 or a signature, read it, then run it. ### EVAL_VAR `eval "$cmd"` executes any `;`, `$(...)` or backticks inside the data. Build commands as arrays: `cmd=(rsync -a -- "$src" "$dst"); "cmd[@]"`. For dispatch, use `case`. ### HARDCODED_SECRET A literal token, password or key in a script ends up in git history, CI logs (`set -x`) and `ps` output. Read it from the environment or a secret manager and fail early: `TOKEN="?"`. Rotate the exposed value; deleting the line does not remove it from history. ## MEDIUM ### NO_STRICT_MODE / NO_PIPEFAIL Without `set -e` a failed command is ignored and the next one runs anyway. Without `pipefail`, `build | tee log` succeeds when build fails. Use `set -euo pipefail` in bash and `set -eu` in POSIX sh. Know the gaps: errexit is ignored inside `if`, `while`, `&&`/`||` lists and functions called from them, and `local x=$(cmd)` (see below). ### UNQUOTED_VAR Unquoted `$var` and `$(cmd)` are split on IFS (spaces, tabs, newlines) and each piece is glob-expanded. `cp $src $dst` with `src="My File.txt"` copies two wrong files; `rm $f` with `f='*'` removes everything. Quote every expansion; use `"$@"` for all arguments and `"arr[@]"` for arrays. Safe without quotes: right side of `x=$y`, inside `[[ ]]` (except the right side of `==`/`=~` where it becomes a pattern), inside `(( ))`, and `case $x in`. ### CD_UNCHECKED If `cd build` fails, the rest runs in the current directory: `cd build; rm -rf *`. Use `cd -- "$dir" || exit 1` or strict mode. In loops, use a subshell: `(cd -- "$d" && make)`. ### LS_PARSING `for f in $(ls *.log)` breaks on spaces and newlines and runs the glob twice. Use the glob directly: `for f in ./*.log; do [ -e "$f" ] || continue; ...; done`, or `find . -name '*.log' -print0 | while IFS= read -r -d '' f; do ...; done`. ### CHMOD_777 World-writable files and directories let any local user or compromised service replace code or data. Use 755/644, 700/600 for private data, or `u=rwX,g=rX,o=`. ### PREDICTABLE_TMP / NO_TRAP_CLEANUP (LOW) A fixed `/tmp/app.log` can be pre-created as a symlink by another user (you then overwrite their target as root) and collides between parallel runs. Use `mktemp`/`mktemp -d` and remove it with `trap '...' EXIT`. ### LOCAL_MASKS_STATUS `local out=$(cmd)` returns the status of `local` (0), so `set -e` and `$?` never see cmd fail. The same goes for `export` and `readonly`. Declare first, assign on the next line. ### SSH_STDIN_LOOP `while read host; do ssh "$host" cmd; done < hosts.txt` stops after the first host because ssh reads the rest of stdin. Use `ssh -n` or `< /dev/null`. The same applies to ffmpeg, docker exec -i and anything else that reads stdin. ### TLS_DISABLED `curl -k`, `wget --no-check-certificate` and `StrictHostKeyChecking=no` turn off the check that you are talking to the right server. Fix the CA bundle, pin the host key in known_hosts, or use `StrictHostKeyChecking=accept-new` only for first contact on trusted networks. ### BASHISM_IN_SH Under `#!/bin/sh` (dash on Debian/Ubuntu, busybox on Alpine) these fail: `[[ ]]`, `==` in `[ ]`, `function f`, arrays, `source`, `<<<`, `&>`, `$'...'`. Switch the shebang to bash or use `[ ]`, `=`, `.`, and positional parameters. ## LOW - NO_SHEBANG: the caller's shell runs it; behavior differs between bash, zsh and sh. - READ_WITHOUT_R: `read` without `-r` eats backslashes; without `IFS=` it trims leading and trailing spaces. - BACKTICKS: hard to nest and quote; use `$(...)`. ## Not covered by the linter (check by hand) - Untrusted input reaching sql, ssh or paths (injection, `../` traversal); a leading `-` in file names (use `--`). - Race conditions on lock files; use `flock`. - Missing `wait` for background jobs, and orphaned jobs on exit. - Locale and tool differences (GNU vs BSD sed -i, date -d, readlink -f on macOS). - Logging secrets with `set -x`; turn xtrace off around secret handling. FILE:references/safe-patterns.md # Safe patterns for bash and sh scripts ## Header ```bash #!/usr/bin/env bash # What it does, how to call it, what it needs (env vars, tools). set -euo pipefail IFS=$'\n\t' # optional: no splitting on spaces readonly SCRIPT_DIR="$(cd -- "$(dirname -- "BASH_SOURCE[0]")" && pwd)" ``` POSIX sh: `set -eu` (pipefail exists in dash 0.5.11+ and busybox, but not everywhere). ## Fail with a message ```bash die() { printf 'error: %s\n' "$*" >&2; exit 1; } command -v jq >/dev/null 2>&1 || die "jq is required" [ -d "$TARGET" ] || die "no such directory: $TARGET" ``` ## Arguments ```bash usage() { echo "usage: $0 [-n] <release>" >&2; exit 2; } dry_run=0 while getopts ":n" opt; do case $opt in n) dry_run=1 ;; *) usage ;; esac done shift $((OPTIND - 1)) release="-"; [ -n "$release" ] || usage ``` ## Temp files and cleanup ```bash tmpdir="$(mktemp -d)" cleanup() { rm -rf -- "$tmpdir"; } trap cleanup EXIT trap 'exit 130' INT TERM # EXIT trap still runs ``` ## Destructive steps ```bash target="?WEB_ROOT is empty" case $target in /|/usr|/etc|"$HOME") die "refusing to clean $target" ;; esac run() { if [ "$dry_run" -eq 1 ]; then echo "+ $*"; else "$@"; fi; } run rm -rf -- "$target"/* ``` ## Loops over files and lines ```bash for f in "$dir"/*.csv; do [ -e "$f" ] || continue; process "$f"; done find "$dir" -type f -name '*.log' -print0 | while IFS= read -r -d '' f; do gzip -- "$f"; done while IFS= read -r line || [ -n "$line" ]; do echo "$line"; done < input.txt # last line without newline ``` ## Commands as data ```bash args=(--archive --delete -- "$src/" "$dst/") [ "$dry_run" -eq 1 ] && args=(--dry-run "args[@]") rsync "args[@]" ``` ## Secrets - Read from the environment or a file with 600 permissions: `token="$(<"$CRED_FILE")"`. - Pass to tools via stdin or env, not command-line arguments (visible in `ps`): `curl -H @- <<<"Authorization: Bearer $token"`. - `set +x` around secret handling. ## Downloads ```bash curl -fsSL --retry 3 -o "$tmpdir/tool.tgz" "$URL" echo "$SHA256 $tmpdir/tool.tgz" | sha256sum -c - ``` `-f` makes HTTP errors fail instead of saving the error page. ## Locking and idempotency ```bash exec 9>/var/lock/deploy.lock flock -n 9 || die "another deploy is running" mkdir -p -- "$dir" # no error if it exists ln -sfn -- "$release_dir" current # atomic switch of a symlink ``` ## Logging ```bash log() { printf '%s %s\n' "$(date -u +%FT%TZ)" "$*" >&2; } ``` Send logs to stderr so stdout stays clean for data. ## Testing a script safely 1. `bash -n script.sh` (syntax only), `shellcheck script.sh`, `python3 scripts/shell_lint.py script.sh`. 2. Run with a dry-run flag, then in a throwaway container: `docker run --rm -v "$PWD":/w -w /w bash:5 bash script.sh`. 3. Trace once with `bash -x script.sh` (secrets off). FILE:templates/shell-review.md # Shell script review: <script name> **Runs from:** <cron on web-01 as root / GitHub Actions ubuntu-24.04 / Docker entrypoint (Alpine, busybox sh)> **Shell:** <bash 5.2 / sh (dash)> **Untrusted inputs:** <$1 release tag, hosts.txt, file names in uploads/> **Verdict:** <BLOCK: 3 HIGH findings / OK with fixes / OK> **Summary:** <one or two sentences: the worst risk and what the fix changes> ## Findings | # | Severity | Code | Line | Problem | Fix | |---|---|---|---|---|---| | 1 | <HIGH> | <RM_EMPTY_VAR> | <15> | <rm -rf $WEB_ROOT/* deletes /* if WEB_ROOT is empty> | <rm -rf -- "?"/*> | ## Manual review notes - <inputs that reach rm/ssh/sql/paths, and how they are validated> - <idempotency: what happens if it runs twice or stops halfway> - <cleanup on exit and on Ctrl+C> ## Fixed script ```bash <full corrected script> ``` ## Key changes ```diff - rm -rf $WEB_ROOT/* + rm -rf -- "?"/* ``` ## Behavior changes - <now exits on the first failed command; previously continued> ## How to test safely 1. `bash -n <script>` and `python3 scripts/shell_lint.py <script>` (expect 0 HIGH) 2. `shellcheck <script>` if available 3. Dry run or a throwaway container: `<command>` 4. <rotate the exposed secret: where and who> FILE:examples/deploy.sh #!/bin/bash # Deploy the web app build to /srv/www and restart the service. # Usage: deploy.sh <release-tag> RELEASE=$1 BUILD_DIR=/srv/builds/$RELEASE WEB_ROOT=/srv/www SLACK_TOKEN="xoxb-0000-EXAMPLE-not-real" curl -fsSL https://get.example-tools.dev/install.sh | sudo bash cd $BUILD_DIR echo "Deploying $RELEASE" > /tmp/deploy.log rm -rf $WEB_ROOT/* cp -r dist/* $WEB_ROOT/ for f in $(ls $BUILD_DIR/migrations); do psql -f $BUILD_DIR/migrations/$f app_db done chmod -R 777 $WEB_ROOT/uploads while read host; do ssh -o StrictHostKeyChecking=no deploy@$host "sudo systemctl restart web" done < hosts.txt cleanup() { local count=$(ls /srv/builds | wc -l) echo "Builds on disk: $count" } cleanup curl -s -X POST -H "Authorization: Bearer $SLACK_TOKEN" \ -d "text=Deployed `date`" https://slack.com/api/chat.postMessage FILE:examples/deploy-fixed.sh #!/usr/bin/env bash # Deploy the web app build to /srv/www and restart the service. # Usage: SLACK_TOKEN=... deploy.sh <release-tag> set -euo pipefail RELEASE="?usage: deploy.sh <release-tag>" BUILD_DIR="/srv/builds/RELEASE" WEB_ROOT="/srv/www" SLACK_TOKEN="?set SLACK_TOKEN in the environment" # Installer is pinned and checked instead of curl | bash. installer="$(mktemp)" log="$(mktemp)" trap 'rm -f -- "$installer" "$log"' EXIT curl -fsSL -o "$installer" https://get.example-tools.dev/install-1.4.2.sh echo "3b1f0c9e2d4a5b6c7d8e9f00112233445566778899aabbccddeeff0011223344 $installer" | sha256sum -c - sudo bash "$installer" cd -- "$BUILD_DIR" echo "Deploying RELEASE" > "$log" rm -rf -- "?"/* cp -r dist/. "$WEB_ROOT/" for f in "$BUILD_DIR"/migrations/*.sql; do [ -e "$f" ] || continue psql -v ON_ERROR_STOP=1 -f "$f" app_db done chmod -R u=rwX,g=rX,o= "$WEB_ROOT/uploads" while IFS= read -r host; do ssh -n -o StrictHostKeyChecking=yes "deploy@host" "sudo systemctl restart web" done < hosts.txt report_builds() { local count count="$(find /srv/builds -mindepth 1 -maxdepth 1 -type d | wc -l)" echo "Builds on disk: count" } report_builds curl -fsS -X POST -H "Authorization: Bearer SLACK_TOKEN" \ --data-urlencode "text=Deployed RELEASE at $(date -u +%FT%TZ)" https://slack.com/api/chat.postMessage FILE:scripts/shell_lint.py #!/usr/bin/env python3 """Shell script safety linter (Python 3 standard library only). Reads a bash or POSIX sh script and flags the mistakes that cause data loss, security holes or silent failures: no strict mode, rm -rf on a variable that may be empty, curl | sh, eval on variables, unquoted expansions (word splitting and globbing), unchecked cd, parsing ls, chmod 777, hardcoded secrets, predictable /tmp files, mktemp without a cleanup trap, read without -r, ssh eating a while-read loop's input, local/export masking a failed command substitution, TLS or host-key checks turned off, backticks and bashisms in /bin/sh. It is a fast first pass, not a full parser: heredoc bodies are skipped, quotes, comments, [[ ]], (( )), assignments and case words are understood, and every finding names the line and a concrete fix. Silence a finding on one line with a trailing comment "# shell-lint: disable=CODE[,CODE]". Usage: python3 shell_lint.py script.sh [more.sh ...] [--json] [--min-severity MEDIUM] [--ignore CODE,CODE] python3 shell_lint.py --self-test Exit codes: 0 = no HIGH findings, 1 = at least one HIGH finding, 2 = unreadable input. """ import argparse, json, re, sys SEV_ORDER = {"HIGH": 0, "MEDIUM": 1, "LOW": 2, "INFO": 3} SECRET_NAME = re.compile(r"[A-Z0-9_]*(PASSWORD|PASSWD|TOKEN|SECRET|API_?KEY|ACCESS_KEY|PRIVATE_KEY)[A-Z0-9_]*") FIXES = { "NO_SHEBANG": "Add '#!/usr/bin/env bash' (or '#!/bin/sh' for POSIX) as line 1 so the right shell runs it.", "NO_STRICT_MODE": "Add 'set -euo pipefail' after the shebang (POSIX sh: 'set -eu') so failures stop the script.", "NO_PIPEFAIL": "Add 'set -o pipefail' so 'cmd | tee log' fails when cmd fails.", "RM_EMPTY_VAR": "Guard the variable: rm -rf -- \"?VAR is empty/\"... and quote it; never build rm paths from unchecked input.", "RM_ROOT": "Never rm the filesystem root or /*; point at a specific directory.", "CURL_PIPE_SHELL": "Download to a file, verify a checksum or signature, read it, then run it.", "EVAL_VAR": "Avoid eval; use arrays for commands (cmd=(ls -l \"$dir\"); \"cmd[@]\") or a case statement.", "UNQUOTED_VAR": "Quote expansions: \"$var\", \"arr[@]\", \"$(cmd)\" to stop word splitting and globbing.", "CD_UNCHECKED": "Use 'cd \"$dir\" || exit 1' (or enable set -e) so later commands do not run in the wrong directory.", "LS_PARSING": "Use a glob (for f in ./*.log) or find -print0 | while IFS= read -r -d '' f instead of parsing ls/find output.", "BACKTICKS": "Use $(...) instead of backticks; it nests and quotes predictably.", "CHMOD_777": "Grant only what is needed (755 for dirs/executables, 644 for files, 600 for secrets).", "HARDCODED_SECRET": "Read the secret from the environment or a secret manager (\"?\"), rotate the exposed value.", "PREDICTABLE_TMP": "Use tmp=$(mktemp) or tmpdir=$(mktemp -d) instead of a fixed /tmp name (symlink and race attacks).", "NO_TRAP_CLEANUP": "Add trap 'rm -rf -- \"$tmpdir\"' EXIT right after mktemp.", "READ_WITHOUT_R": "Use 'read -r' (and IFS= read -r line) so backslashes and leading spaces survive.", "LOCAL_MASKS_STATUS": "Split it: 'local out; out=$(cmd)' so set -e and $? see the command's failure.", "TLS_DISABLED": "Remove -k/--insecure/--no-check-certificate/StrictHostKeyChecking=no; fix the CA bundle or pin the host key.", "SSH_STDIN_LOOP": "Use 'ssh -n' (or ssh ... < /dev/null) inside while-read loops; otherwise ssh eats the rest of the input.", "BASHISM_IN_SH": "Either change the shebang to bash or use the POSIX form ([ ], =, . file, no arrays).", } class Finding: def __init__(self, sev, code, line, snippet, problem): self.sev, self.code, self.line, self.snippet, self.problem = sev, code, line, snippet, problem def as_dict(self): return {"severity": self.sev, "code": self.code, "line": self.line, "snippet": self.snippet, "problem": self.problem, "fix": FIXES.get(self.code, "")} def strip_comment(line): """Remove an unquoted comment. '#' starts a comment only at the start of a word.""" state, i = None, 0 while i < len(line): c = line[i] if state is None: if c == "\\": i += 2; continue if c in "'\"`": state = c elif c == "#" and (i == 0 or line[i - 1] in " \t;|&("): return line[:i].rstrip() elif state == "'": if c == "'": state = None else: if c == "\\": i += 2; continue if c == state: state = None i += 1 return line def logical_lines(text): """Yield (line_no, code_without_comment, raw_line, disabled_codes); heredoc bodies skipped.""" lines = text.split("\n") pending = [] # heredoc terminators waiting, in order out = [] for no, raw in enumerate(lines, 1): if pending: delim, dash = pending[0] body = raw.lstrip("\t") if dash else raw if body == delim: pending.pop(0) continue m = re.search(r"#\s*shell-lint:\s*disable=([A-Z_,\s]+)", raw) disabled = {c.strip() for c in m.group(1).split(",")} if m else set() code = strip_comment(raw) for hm in re.finditer(r"<<(-?)\s*(['\"]?)([A-Za-z_][A-Za-z0-9_]*)\2", code): if code[hm.start():hm.start() + 3] == "<<<": continue pending.append((hm.group(3), hm.group(1) == "-")) out.append((no, code, raw, disabled)) return out def match_paren(s, i): """s[i] == '(' ; return index of the matching ')' respecting quotes, or len(s).""" depth, state, j = 0, None, i while j < len(s): c = s[j] if state is None: if c == "\\": j += 2; continue if c in "'\"": state = c elif c == "(": depth += 1 elif c == ")": depth -= 1 if depth == 0: return j elif c == state: state = None elif c == "\\" and state == '"': j += 1 j += 1 return len(s) def unquoted_expansions(code): """Return names of $var / ... / $(...) / $@ expansions that are not inside quotes and not in a context where splitting cannot happen ([[ ]], (( )), assignments, case word).""" found = [] stripped = code.strip() if re.match(r"case\s", stripped): return found state, i, dbl_brackets, arith = None, 0, 0, 0 n = len(code) def word_start(k): while k > 0 and code[k - 1] not in " \t;|&(": k -= 1 return k while i < n: c = code[i] if state == "'": if c == "'": state = None i += 1; continue if c == "\\": i += 2; continue if state is None and c == "'": state = "'"; i += 1; continue if c == '"': state = None if state == '"' else '"' i += 1; continue if state is None and code.startswith("[[", i) and (i == 0 or code[i - 1] in " \t;&|!("): dbl_brackets += 1; i += 2; continue if state is None and code.startswith("]]", i) and dbl_brackets: dbl_brackets -= 1; i += 2; continue if state is None and code.startswith("((", i) and (i == 0 or code[i - 1] in " \t;&|!"): end = match_paren(code, i) i = end + 1; continue if c == "`": end = code.find("`", i + 1) i = (end if end != -1 else n) + 1; continue if c == "$" and i + 1 < n: nxt = code[i + 1] if code.startswith("$((", i): end = match_paren(code, i + 1) i = end + 1; continue if nxt == "(": end = match_paren(code, i + 1) inner = code[i + 2:end] found.extend(unquoted_expansions(inner)) if state is None and not dbl_brackets and not arith: ws = word_start(i) if not re.match(r"[A-Za-z_][A-Za-z0-9_]*(\[[^\]]*\])?\+?=", code[ws:i + 1]): found.append("$(" + inner.strip()[:20] + ")") i = end + 1; continue m = re.match(r"\$(\{[^}]*\}|[A-Za-z_][A-Za-z0-9_]*|[@*]|[0-9])", code[i:]) if m: tok = m.group(0) if state is None and not dbl_brackets: ws = word_start(i) is_assign = re.match(r"[A-Za-z_][A-Za-z0-9_]*(\[[^\]]*\])?\+?=", code[ws:i + 1]) if not is_assign and not tok.startswith("found.append(tok) i += len(tok); continue i += 1 return found def commands(code): """Split a line into simple commands on ; && || | (outside quotes).""" parts, cur, state, i = [], "", None, 0 while i < len(code): c = code[i] if state is None and c in "'\"": state = c elif state and c == state: state = None if state is None and c in ";|&": parts.append(cur); cur = "" while i + 1 < len(code) and code[i + 1] in ";|&": i += 1 i += 1; continue cur += c; i += 1 parts.append(cur) return [p.strip() for p in parts if p.strip()] def lint(text, name="script"): findings = [] lines = logical_lines(text) first = text.split("\n", 1)[0] shebang = first if first.startswith("#!") else "" is_posix_sh = bool(re.search(r"(^#!\s*/bin/sh\b|/env\s+sh\b|/bin/dash\b)", shebang)) all_code = "\n".join(c for _, c, _, _ in lines) strict_e = bool(re.search(r"(^|\n)\s*set\s+(-[a-zA-Z]*e[a-zA-Z]*\b|-o\s+errexit)", all_code)) or bool(re.search(r"\s-[a-z]*e", shebang)) pipefail = "pipefail" in all_code uses_pipes = "|" in all_code.replace("||", "") uses_mktemp = "mktemp" in all_code has_trap = bool(re.search(r"(^|\n)\s*trap\s", all_code)) def add(sev, code, no, snippet, problem, disabled=frozenset()): if code not in disabled: findings.append(Finding(sev, code, no, snippet.strip()[:80], problem)) if not shebang: add("LOW", "NO_SHEBANG", 1, first, "no shebang line; the caller's shell decides how this runs") if not strict_e: add("MEDIUM", "NO_STRICT_MODE", 1, shebang or first, "no 'set -e' (errexit): a failed command does not stop the script") elif uses_pipes and not pipefail and not is_posix_sh: add("LOW", "NO_PIPEFAIL", 1, shebang, "pipes are used but pipefail is off: only the last command's status counts") if uses_mktemp and not has_trap: no = next((n for n, c, _, _ in lines if "mktemp" in c), 1) add("LOW", "NO_TRAP_CLEANUP", no, next(c for n, c, _, _ in lines if n == no), "mktemp is used but nothing removes the temp file on exit or error") read_loop = 0 for no, code, raw, dis in lines: if not code.strip(): continue if re.search(r"(^|[;&|\s])while\b.*\bread\b", code): read_loop += 1 if read_loop and re.search(r"(^|[;&|\s])ssh\s", code) and not re.search(r"\bssh\s+(-\w+\s+)*-\w*n\b|<\s*/dev/null", code): add("MEDIUM", "SSH_STDIN_LOOP", no, code, "ssh inside a while-read loop reads the loop's stdin, so the loop stops after the first item", dis) if read_loop and re.search(r"(^|[;&|\s])done\b", code): read_loop -= 1 # curl | sh if re.search(r"\b(curl|wget)\b[^|]*\|\s*(sudo\s+(-\S+\s+)*)?(ba|z|da|k)?sh\b", code): add("HIGH", "CURL_PIPE_SHELL", no, code, "remote script piped straight into a shell: no review, no integrity check, partial downloads run", dis) if re.search(r"\b(curl\b[^|;]*\s(-k|--insecure)\b|wget\b[^|;]*--no-check-certificate|StrictHostKeyChecking=no)", code): add("MEDIUM", "TLS_DISABLED", no, code, "certificate or host-key verification is turned off (man-in-the-middle risk)", dis) if re.search(r"(^|[;&|(\s])eval\s+.*\$", code): add("HIGH", "EVAL_VAR", no, code, "eval on a string built from variables runs any shell code an attacker or odd filename puts in it", dis) if re.search(r"\bchmod\s+(-[A-Za-z]+\s+)*0?777\b|\bchmod\s+(-[A-Za-z]+\s+)*[augo]*\+rwx\b", code) and re.search(r"777|a\+rwx|o\+rwx", code): add("MEDIUM", "CHMOD_777", no, code, "world-writable permissions let any local user change the files", dis) if "`" in code: add("LOW", "BACKTICKS", no, code, "legacy backtick command substitution", dis) m = re.match(r"\s*(export\s+|readonly\s+|local\s+|declare\s+(-\w+\s+)*)?(" + SECRET_NAME.pattern + r")=(['\"]?)([^$'\"\s][^'\"\s]{3,)", code) if m: val = m.group(6) masked = val[:2] + "*" * max(3, len(val) - 4) + val[-2:] add("HIGH", "HARDCODED_SECRET", no, code.replace(val, masked), f"{m.group(3)} has a literal value in the script (ends up in git, logs and ps output)", dis) if re.match(r"\s*(local|export|declare|readonly)\s+(-\w+\s+)*[A-Za-z_][A-Za-z0-9_]*=\"?\$\(", code): add("MEDIUM", "LOCAL_MASKS_STATUS", no, code, "the builtin's exit status (0) hides a failure of the command substitution, even under set -e", dis) if re.search(r"(^|[^\w.-])/tmp/[A-Za-z0-9_.${}-]+", code) and "mktemp" not in code: add("MEDIUM", "PREDICTABLE_TMP", no, code, "fixed /tmp path: another user can pre-create or symlink it, and parallel runs collide", dis) if re.search(r"\bfor\s+\w+\s+in\s+(\$\(|`)\s*(ls|find)\b", code) or re.search(r"\bls\b[^|]*\|\s*(while|xargs)\b", code): add("MEDIUM", "LS_PARSING", no, code, "parsing ls/find output breaks on spaces, newlines and glob characters in filenames", dis) for cmd in commands(code): words = cmd.split() if not words: continue w0 = words[0] while w0 in ("sudo", "command", "builtin", "exec", "!", "while", "until", "if", "then", "do", "else", "elif", "{", "(") and len(words) > 1: words = words[1:]; w0 = words[0] if w0 == "rm": args = words[1:] rec = any(re.match(r"-[a-zA-Z]*[rR]", a) or a == "--recursive" for a in args) if any(a.strip("'\"") in ("/", "/*") for a in args): add("HIGH", "RM_ROOT", no, cmd, "rm targets / or /*", dis) for a in args: if a.startswith("-") and a != "--": continue if "$" in a and not re.search(r"\$\{[A-Za-z_][A-Za-z0-9_]*:\?", a): a_clean = a.strip("'\"") if re.match(r"\$\{?[A-Za-z_][A-Za-z0-9_]*\}?(/|\"?$)", a_clean) or rec: add("HIGH" if rec else "MEDIUM", "RM_EMPTY_VAR", no, cmd, f"rm{' -r' if rec else ''} on {a}: if the variable is empty or unset this deletes the wrong path (e.g. '/')", dis) break if w0 == "cd" and not strict_e: if not re.search(re.escape(cmd) + r"\s*(\|\||&&)", code) and not re.search(r"(\|\||&&)\s*" + re.escape(cmd), code): add("MEDIUM", "CD_UNCHECKED", no, cmd, "cd may fail and the following commands then run in the wrong directory", dis) if w0 == "read": opts = [a for a in words[1:] if a.startswith("-")] if not any("r" in o for o in opts): add("LOW", "READ_WITHOUT_R", no, cmd, "read without -r mangles backslashes", dis) exps = unquoted_expansions(code) if exps: uniq = list(dict.fromkeys(exps)) shown = ", ".join(uniq[:4]) + (" ..." if len(uniq) > 4 else "") only_echo = re.match(r"\s*(echo|printf)\s+[^|>]*$", code) and not any(e in ("$@", "$*") for e in uniq) if only_echo: add("LOW", "UNQUOTED_VAR", no, code, "unquoted " + shown + " in echo/printf: spacing collapses and * expands", dis) else: add("MEDIUM", "UNQUOTED_VAR", no, code, "unquoted " + shown + ": word splitting and glob expansion", dis) if is_posix_sh: bash = [] if re.search(r"(^|[\s;&|!])\[\[\s", code): bash.append("[[ ]]") if re.search(r"(^|[\s;&|])\[\s[^]]*\s==\s", code): bash.append("== in [ ]") if re.match(r"\s*function\s+\w+", code): bash.append("function keyword") if re.search(r"\b[A-Za-z_]\w*=\(", code): bash.append("array") if re.match(r"\s*source\s", code): bash.append("source") if "<<<" in code: bash.append("here-string") if re.search(r"&>", code): bash.append("&>") if re.search(r"\$'", code): bash.append("$'...'") if bash: add("MEDIUM", "BASHISM_IN_SH", no, code, "bash-only syntax under #!/bin/sh (" + ", ".join(bash) + "); dash and busybox sh will fail", dis) findings.sort(key=lambda f: (SEV_ORDER[f.sev], f.line, f.code)) return findings def render_md(name, findings): counts = {s: sum(1 for f in findings if f.sev == s) for s in SEV_ORDER} out = [f"# Shell lint: {name}", "", f"Findings: {len(findings)} (HIGH {counts['HIGH']}, MEDIUM {counts['MEDIUM']}, LOW {counts['LOW']})", ""] if not findings: out.append("No findings.") return "\n".join(out) out += ["| Severity | Code | Line | Snippet | Problem |", "|---|---|---|---|---|"] for f in findings: snip = f.snippet.replace("|", "\\|").replace("`", "'") out.append(f"| {f.sev} | {f.code} | {f.line} | `{snip}` | {f.problem.replace('|', '/')} |") out += ["", "## Fixes", ""] for code in dict.fromkeys(f.code for f in findings): out.append(f"- {code}: {FIXES[code]}") return "\n".join(out) # ---------------------------------------------------------------- self-test def _codes(src): return [f.code for f in lint(src)] def self_test(): S = "#!/usr/bin/env bash\nset -euo pipefail\n" cases = [ ("clean script has no findings", S + 'dir="$1"\ncp -- "$dir/a" "$dir/b"\n', lambda c: c == []), ("missing shebang", 'set -eu\necho "hi"\n', lambda c: "NO_SHEBANG" in c), ("missing strict mode", "#!/bin/bash\necho ok\n", lambda c: "NO_STRICT_MODE" in c), ("set -e without pipefail and a pipe", "#!/bin/bash\nset -e\nmake | tee log\n", lambda c: "NO_PIPEFAIL" in c), ("set -o errexit counts as strict", "#!/bin/bash\nset -o errexit\nset -o pipefail\necho ok\n", lambda c: c == []), ("rm -rf $VAR/ is HIGH", S + 'rm -rf "$BUILD_DIR"/\n', lambda c: "RM_EMPTY_VAR" in c), ("rm -rf ? is fine", S + 'rm -rf -- "?/"\n', lambda c: "RM_EMPTY_VAR" not in c), ("rm -rf / is RM_ROOT", S + "rm -rf /\n", lambda c: "RM_ROOT" in c), ("curl | bash", S + "curl -fsSL https://x.example/install.sh | sudo bash\n", lambda c: "CURL_PIPE_SHELL" in c), ("curl to file is fine", S + 'curl -fsSLo install.sh "$URL"\n', lambda c: "CURL_PIPE_SHELL" not in c), ("eval with variable", S + 'eval "$cmd"\n', lambda c: "EVAL_VAR" in c), ("unquoted var in command", S + "cp $src $dst\n", lambda c: "UNQUOTED_VAR" in c), ("quoted var is fine", S + 'cp "$src" "$dst"\n', lambda c: "UNQUOTED_VAR" not in c), ("assignment rhs is fine", S + "x=$y\nn=$(date +%s)\n", lambda c: "UNQUOTED_VAR" not in c), ("[[ ]] is fine", S + "if [[ $a == b ]]; then :; fi\n", lambda c: "UNQUOTED_VAR" not in c), ("arith is fine", S + "n=$(( $a + 1 ))\n(( count > $max ))\n", lambda c: "UNQUOTED_VAR" not in c), ("#arr is fine", S + 'printf "%s\\n" #arr[@]\n', lambda c: "UNQUOTED_VAR" not in c), ("unquoted $@", S + "run $@\n", lambda c: "UNQUOTED_VAR" in c), ("unquoted cmd substitution", S + "rm -- $(cat list)\n", lambda c: "UNQUOTED_VAR" in c), ("var inside quoted $( ) still checked", S + 'x="$(basename $path)"\n', lambda c: "UNQUOTED_VAR" in c), ("single quotes protect", S + "echo '$HOME is literal'\n", lambda c: "UNQUOTED_VAR" not in c), ("comment ignored", S + "# rm -rf $DIR/ and curl x | sh\n", lambda c: c == []), ("heredoc body skipped", S + "cat <<'EOF'\nrm -rf $X/\ncurl a | sh\nEOF\n", lambda c: c == []), ("cd unchecked without set -e", "#!/bin/bash\ncd /srv/app\n", lambda c: "CD_UNCHECKED" in c), ("cd || exit is fine", "#!/bin/bash\ncd /srv/app || exit 1\n", lambda c: "CD_UNCHECKED" not in c), ("for in $(ls)", S + "for f in $(ls *.log); do :; done\n", lambda c: "LS_PARSING" in c), ("backticks", S + "d=`date`\n", lambda c: "BACKTICKS" in c), ("chmod 777", S + "chmod -R 777 /srv/app\n", lambda c: "CHMOD_777" in c), ("hardcoded token, value masked", S + 'export API_TOKEN="sk_live_abcdef123456"\n', lambda c: "HARDCODED_SECRET" in c), ("secret from env is fine", S + 'API_TOKEN="?"\n', lambda c: "HARDCODED_SECRET" not in c), ("predictable tmp", S + 'echo "x" > /tmp/deploy.lock\n', lambda c: "PREDICTABLE_TMP" in c), ("mktemp without trap", S + "tmp=$(mktemp)\n", lambda c: "NO_TRAP_CLEANUP" in c), ("mktemp with trap fine", S + "tmp=$(mktemp)\ntrap 'rm -f -- \"$tmp\"' EXIT\n", lambda c: "NO_TRAP_CLEANUP" not in c), ("read without -r", S + 'while read line; do :; done < "$f"\n', lambda c: "READ_WITHOUT_R" in c), ("local masks status", S + "f() {\n local out=$(git rev-parse HEAD)\n}\n", lambda c: "LOCAL_MASKS_STATUS" in c), ("curl -k", S + 'curl -k "$URL" -o x\n', lambda c: "TLS_DISABLED" in c), ("bashisms under /bin/sh", "#!/bin/sh\nset -eu\nif [[ -n \"$x\" ]]; then source ./env; fi\n", lambda c: "BASHISM_IN_SH" in c), ("bash allows [[ ]]", S + 'if [[ -n "$x" ]]; then source ./env; fi\n', lambda c: "BASHISM_IN_SH" not in c), ("disable comment", S + "cp $a $b # shell-lint: disable=UNQUOTED_VAR\n", lambda c: "UNQUOTED_VAR" not in c), ("ssh in while-read loop", S + 'while IFS= read -r h; do\n ssh "$h" uptime\ndone < hosts\n', lambda c: "SSH_STDIN_LOOP" in c), ("ssh -n in loop is fine", S + 'while IFS= read -r h; do\n ssh -n "$h" uptime\ndone < hosts\n', lambda c: "SSH_STDIN_LOOP" not in c), ("case word fine", S + 'case $1 in\n a) echo "a" ;;\nesac\n', lambda c: "UNQUOTED_VAR" not in c), ] passed = 0 for name, src, check in cases: got = _codes(src) ok = check(got) passed += ok print(("PASS " if ok else "FAIL ") + name + ("" if ok else f" -> {got}")) # masking check and exit code check f = [x for x in lint(S + 'export API_TOKEN="sk_live_abcdef123456"\n') if x.code == "HARDCODED_SECRET"][0] ok = "abcdef123456" not in f.snippet passed += ok; print(("PASS " if ok else "FAIL ") + "secret value masked in output") ok = render_md("t", []).endswith("No findings.") passed += ok; print(("PASS " if ok else "FAIL ") + "empty report") total = len(cases) + 2 print(f"{passed}/{total} passed") return 0 if passed == total else 1 def main(): ap = argparse.ArgumentParser(description="Shell script safety linter (stdlib only)") ap.add_argument("files", nargs="*") ap.add_argument("--json", action="store_true") ap.add_argument("--min-severity", default="LOW", choices=list(SEV_ORDER)) ap.add_argument("--ignore", default="", help="comma-separated codes to skip") ap.add_argument("--self-test", action="store_true") a = ap.parse_args() if a.self_test: return self_test() if not a.files: ap.error("give at least one script or --self-test") ignore = {c.strip() for c in a.ignore.split(",") if c.strip()} worst, reports, js = 0, [], {} for path in a.files: try: with open(path, encoding="utf-8") as fh: text = fh.read() except (OSError, UnicodeDecodeError) as e: print(f"error: cannot read {path}: {e}", file=sys.stderr) return 2 fs = [f for f in lint(text, path) if f.code not in ignore and SEV_ORDER[f.sev] <= SEV_ORDER[a.min_severity]] if any(f.sev == "HIGH" for f in fs): worst = 1 js[path] = [f.as_dict() for f in fs] reports.append(render_md(path, fs)) print(json.dumps(js, indent=2) if a.json else "\n\n".join(reports)) return worst if __name__ == "__main__": sys.exit(main())
审查 .service 与 drop-in：禁止 root 乱跑、Environment 泄密、权限过宽等。
---
name: systemd-service-unit-hardening-reviewer
description: 审查 .service 与 drop-in：禁止 root 乱跑、Environment 泄密、权限过宽等。
---
# systemd 服务单元加固审查
你是「systemd 服务单元加固审查」技能助手。上方 description 为中文任务说明。下方保留原技能完整正文（含脚本与代码）；执行时遵循中文说明与原文技术细节。
# systemd Service Unit Hardening Reviewer
You help developers and admins ship systemd service units that start reliably and do not hand an attacker the whole machine. Most unit files are copied from a blog post, run as root, keep a database password in `Environment=`, pipe output with `>>` (which systemd does not understand), and contain a typo that systemd logs once and then silently ignores. Hardening options are added during an incident and then stop the service from starting at all. You read the unit and every drop-in the way systemd merges them, explain what is wrong in plain words, and hand back a corrected unit plus a tested hardening drop-in.
You review the text you are given. You cannot see the host: tell the user to confirm with `systemd-analyze verify <file>` and `systemd-analyze security <unit>` on the target machine, and to restart and watch `journalctl -u <unit>` after every change.
## Files in this skill
- `scripts/systemd_unit_review.py` - parses a .service file plus drop-in .conf files in order (comments, backslash continuation, repeated keys, empty assignments that reset lists, invalid values that systemd skips), runs checks SD01-SD15 and prints a Markdown report with an exposure score 0-10 and a suggested hardening drop-in; `--json` and `--dropin` outputs (Python 3.8+ standard library only)
- `references/directives-and-sandboxing.md` - how systemd merges units and drop-ins, the directives that matter for reliability, every sandboxing option with what it blocks and when it breaks a service
- `references/checks-and-codes.md` - every finding code, its severity and the fix
- `templates/unit-review.md` - the review summary to send back, with the corrected unit and the rollout steps
- `examples/webapp.service` + `examples/webapp-override.conf` - a hand-deployed Node.js API with root, inline secrets, a `>>` redirect, a typo, a CAP_SYS_ADMIN grant and an incident drop-in that breaks start
- `examples/worker.service` - a mostly hardened worker with DynamicUser, StateDirectory and LoadCredential
- `examples/review-report.md` - the real checker output on both units
## Workflow
### 1. Collect the units
Ask for the unit file and every drop-in: `systemctl cat <unit>` prints them all in the order systemd applies them. Ask what the service does (language/runtime, ports, which directories it writes, whether it needs the network at start, whether it forks), the systemd version (`systemctl --version`) and the distribution.
### 2. Run the checker
```
python3 scripts/systemd_unit_review.py examples/webapp.service examples/webapp-override.conf examples/worker.service
python3 scripts/systemd_unit_review.py myapp.service override.conf --dropin
python3 scripts/systemd_unit_review.py myapp.service --json
python3 scripts/systemd_unit_review.py --self-test
```
Files after a .service are its drop-ins until the next .service. Exit 0 = no high findings, 1 = high findings, 2 = bad input. Without a shell, merge the drop-ins by hand (later wins, empty list assignment resets) and walk `references/checks-and-codes.md`.
### 3. Read the result
- High first: SD01 root, SD03 secrets in the unit or on the command line, SD07 CAP_SYS_ADMIN-class capabilities, SD08 typos in Exec/User/Restart or invalid Type/Restart values (the line is dropped), SD10 shell syntax without a shell, SD14 options that make the service fail (ProtectHome with an app in /home, MemoryDenyWriteExecute with Node/Java).
- Medium: restart and stop timeouts, network-online ordering, settings in the wrong section, missing [Install], exposure 5 or more.
- Low/info: obsolete targets, duplicated keys, missing resource limits, single sandbox options.
### 4. Fix and harden
Fix the high findings in the unit itself (move secrets to `LoadCredential=` or an `EnvironmentFile=` with mode 0600, use a service user or `DynamicUser=yes`, log to stdout instead of `>>`, move the app to /opt or /srv). Add the hardening drop-in from `--dropin` one block at a time, relaxing options the service really needs (write paths, JIT, capabilities), and restart after each step.
### 5. Deliver
Fill `templates/unit-review.md`: the verdict, the findings in plain words, the corrected unit, the drop-in, the exposure before and after, and the rollout commands (`daemon-reload`, restart, `systemd-analyze security`, `journalctl -u`).
## Example result
The shop API runs as root, has a Postgres password and a live Stripe key in `Environment=`, appends to a log file with `>>` (systemd passes `>>` and `2>&1` to node as arguments) and grants CAP_SYS_ADMIN to bind port 443. `RestartSecs=5` is a typo, so restarts come every 100 ms. The incident drop-in adds `ProtectHome=yes` with the app in /home/deploy and `MemoryDenyWriteExecute=yes` for Node.js - both stop it from starting - and its `Restart=on-faliure` is ignored. 9 high findings, exposure 8.1. The worker is in good shape (exposure 1.2, six optional options to add).
## Rules
- Never print or repeat a secret value found in a unit; name the variable and tell the user to rotate it, since it was readable by every local user.
- Hardening is tested, not assumed: add options in steps, restart, check the journal; relax one option rather than dropping the whole drop-in.
- Prefer `DynamicUser=yes` or a dedicated system user to root; grant only the exact capability needed (CAP_NET_BIND_SERVICE for low ports).
- Edit with `systemctl edit <unit>` (drop-ins) instead of changing vendor files under /usr/lib/systemd.
- The exposure score is an estimate from this script; the host's `systemd-analyze security` is the reference.
FILE:references/directives-and-sandboxing.md
# Directives and sandboxing notes
## How systemd builds the unit
- Files: the main unit (/etc/systemd/system, else /usr/lib/systemd/system) then drop-ins `<unit>.d/*.conf` in name order. `systemctl cat <unit>` shows them all.
- Later assignments win for single-value keys. List keys (After=, Environment=, ExecStartPre=, ReadWritePaths= ...) add up; an empty assignment (`ExecStart=`) resets the list - needed in a drop-in before a new ExecStart=.
- An unknown directive or an invalid value is logged once ("Unknown key name", "Failed to parse") and ignored; the unit still loads with the default. That is why typos survive for years.
- Lines ending in `\` continue on the next line. `#` and `;` start comments only at the start of a line.
- `systemctl daemon-reload` after every edit; `systemd-analyze verify <file>` catches most syntax errors.
## Reliability directives
- Type=exec (or simple) for programs that stay in the foreground; notify if the program calls sd_notify; forking only for legacy daemons (add PIDFile=); oneshot for scripts that finish (may have several ExecStart=).
- Restart=on-failure + RestartSec=5s for most services. Restart=always also restarts after a clean exit. Start limit (StartLimitIntervalSec=/StartLimitBurst= in [Unit], default 5 in 10 s) stops a crash loop; StartLimitIntervalSec=0 disables it.
- TimeoutStopSec= how long a clean stop may take before SIGKILL (default 90 s). infinity hangs shutdowns.
- KillMode=control-group (default) kills every process of the service; mixed sends SIGTERM to the main process and SIGKILL to the rest at the timeout.
- Network: `Wants=network-online.target` + `After=network-online.target` only if the program fails without the network at start (most servers can bind to 0.0.0.0 without it).
- Logs: write to stdout/stderr; journald stores them (`journalctl -u <unit>`). Use LogsDirectory= if a file is required.
- [Install] WantedBy=multi-user.target so `systemctl enable` starts it at boot.
## Exec lines are not shell
systemd splits the line into words itself: `|`, `>`, `>>`, `2>&1`, `&&`, `;`, `&`, `$(...)` and backticks are passed to the program as arguments. `$VAR` and `VAR` are expanded from Environment=. If you need a shell use `ExecStart=/bin/sh -c '...'`, but a small script is easier to read and test. Prefixes: `-` ignore failure, `@` argv0, `+` full privileges, `!` privileges without the user switch (both skip the sandbox).
## Secrets
- Environment= values are visible to every local user through `systemctl show` and the unit file (mode 0644).
- `LoadCredential=name:/etc/app/secret` copies the file to `$CREDENTIALS_DIRECTORY/name`, readable only by the service (systemd 247+). `SetCredentialEncrypted=` / `systemd-creds` for encrypted ones (250+).
- EnvironmentFile=/etc/app/env with owner root and mode 0600 is the portable fallback.
- Command-line secrets are visible in `ps` and /proc/<pid>/cmdline.
## Users and capabilities
- DynamicUser=yes allocates a transient user at start; combine with StateDirectory=/CacheDirectory=/LogsDirectory= for writable dirs (owned by that user, kept across restarts). Implies ProtectSystem=strict, ProtectHome=read-only, PrivateTmp.
- Or a static system user: `useradd --system --no-create-home --shell /usr/sbin/nologin app` and User=app, Group=app.
- Port below 1024 as non-root: `AmbientCapabilities=CAP_NET_BIND_SERVICE` and `CapabilityBoundingSet=CAP_NET_BIND_SERVICE`. CAP_SYS_ADMIN is almost root - never for a web app.
## Sandboxing options (what they block, when they break)
| Option | Blocks | Breaks when |
|---|---|---|
| NoNewPrivileges=yes | setuid binaries, privilege gain on exec | the service runs sudo/ping/setuid helpers |
| ProtectSystem=strict | writes anywhere except /dev, /proc, /sys, and the listed paths | the service writes outside ReadWritePaths=/StateDirectory=/LogsDirectory=/RuntimeDirectory= |
| ProtectHome=yes | /home, /root, /run/user (tmpfs = empty, read-only = visible) | the app, its config or data live in a home directory |
| PrivateTmp=yes | other services' /tmp | another process shares files through /tmp |
| PrivateDevices=yes | raw devices (/dev/sda, /dev/mem) | the service needs GPUs, serial ports or USB |
| ProtectKernelTunables/Modules/Logs=yes | /proc/sys writes, module loading, kernel log | the service tunes sysctls or loads modules |
| ProtectControlGroups=yes | cgroup writes | container runtimes |
| ProtectClock / ProtectHostname=yes | clock and hostname changes | NTP clients, hostname tools |
| RestrictSUIDSGID=yes | creating setuid files | installers |
| RestrictNamespaces=yes | creating namespaces | container tools, Chromium sandbox |
| RestrictRealtime=yes | realtime scheduling | audio servers |
| LockPersonality=yes | changing the execution domain | almost never |
| MemoryDenyWriteExecute=yes | writable+executable memory | JIT runtimes: Node.js, Java, .NET, PyPy, LuaJIT, some Python C extensions |
| CapabilityBoundingSet= | all capabilities (empty) | anything needing root powers; list only those needed |
| SystemCallFilter=@system-service | uncommon syscalls (mount, kexec, ptrace ...) | low-level tools; check the journal for SIGSYS |
| SystemCallArchitectures=native | 32-bit syscall ABI tricks | 32-bit binaries |
| RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 | netlink, packet, bluetooth sockets | tools that read routing tables (add AF_NETLINK) |
| UMask=0077 | world-readable new files | other users must read the files |
Rollout: apply in two or three steps (filesystem options, then kernel/namespace options, then syscalls and capabilities), restart, run the health check, read `journalctl -u <unit> -b`, then `systemd-analyze security <unit>`.
FILE:references/checks-and-codes.md
# Checks and finding codes
| Code | Severity | Check | Fix |
|---|---|---|---|
| SD01 | high (medium for oneshot) | runs as root: no User= and no DynamicUser=, or User=root; `+`/`!` Exec prefixes bypass the user and sandbox (medium) | DynamicUser=yes or a system user; drop the prefix |
| SD02 | high: no ExecStart, several ExecStart for a non-oneshot; low: command not an absolute path | ExecStart sanity | one absolute ExecStart; empty `ExecStart=` first in a drop-in to replace it |
| SD03 | high | secret-looking variable (PASSWORD, SECRET, TOKEN, API_KEY, PRIVATE_KEY, DSN) or a URL with user:password in Environment=; `--password=` style secrets on the command line | LoadCredential= or EnvironmentFile= with mode 0600; rotate the leaked secret |
| SD04 | medium: Restart=no for a long-running service; low: no RestartSec (100 ms default); high: oneshot with Restart=always/on-success; info: Restart=always with the default start limit | restart policy | Restart=on-failure, RestartSec=5s |
| SD05 | medium: After= and Wants= of network-online.target not paired; low: obsolete syslog.target / network.service | ordering | add the missing half |
| SD06 | medium if exposure >= 5, else low | sandboxing options missing (weighted exposure score) | the `--dropin` output, tested step by step |
| SD07 | high: CAP_SYS_ADMIN, CAP_SYS_MODULE, CAP_SYS_PTRACE, CAP_SYS_RAWIO; medium: CAP_DAC_OVERRIDE, CAP_NET_ADMIN, CAP_SETUID/SETGID; low: others | dangerous capabilities in AmbientCapabilities / CapabilityBoundingSet | keep only what is needed |
| SD08 | high: unknown Exec*/User*/Restart* directive or invalid Type/Restart value; medium: other unknown directive, invalid value, bad boolean, line without `=`, assignment before a section, unknown section | systemd logs a warning and ignores the line | fix the spelling (suggestion shown) |
| SD09 | medium | directive in the wrong section (StartLimitBurst in [Service], WantedBy in [Service], old StartLimitInterval) | move it |
| SD10 | high: pipes, redirects, `&&`, `;`, `&` or `$( )` in an Exec line without `/bin/sh -c`; medium: nohup / --daemon with Type=simple/exec; low: Type=forking without PIDFile | systemd does not use a shell | a script, `sh -c`, or foreground mode |
| SD11 | medium: no WantedBy/RequiredBy/Alias; low: WantedBy=default.target or graphical.target for a system service | enable at boot | WantedBy=multi-user.target |
| SD12 | medium: TimeoutStopSec=infinity/0, KillMode=none; low: KillMode=process | stop behaviour | TimeoutStopSec=30s, KillMode=mixed |
| SD13 | low | the same single-value key twice in one file | keep one |
| SD14 | high: ProtectHome=yes/tmpfs with an app, working dir or env file under /home or /root; MemoryDenyWriteExecute with node/java/dotnet/pypy/bun/deno; medium: ProtectSystem=strict with a write path not in ReadWritePaths/StateDirectory/LogsDirectory/RuntimeDirectory | options that break the service | move the app, add the write path, drop the option |
| SD15 | info | no MemoryMax/MemoryHigh/TasksMax/CPUQuota | add limits from observed usage |
## Exposure score
Weights (sum 10): run as non-root 2.0, ProtectSystem=strict 1.0 (full/yes counts half), NoNewPrivileges 0.8, ProtectHome 0.6, CapabilityBoundingSet without dangerous caps 0.6, PrivateTmp 0.5, PrivateDevices 0.5, ProtectKernelTunables 0.4, ProtectKernelModules 0.4, RestrictNamespaces 0.4, SystemCallFilter 0.4, RestrictAddressFamilies 0.4, ProtectControlGroups 0.3, RestrictSUIDSGID 0.3, MemoryDenyWriteExecute 0.3, ProtectKernelLogs 0.2, ProtectClock 0.2, RestrictRealtime 0.2, UMask 0077/0027 0.2, ProtectHostname 0.1, LockPersonality 0.1, SystemCallArchitectures=native 0.1.
Rating: under 2 OK, under 5 MEDIUM, under 8 EXPOSED, 8 or more UNSAFE.
## Verdict
- **N high finding(s) - fix before deploying**
- **works, but exposure X - add the hardening drop-in** (no high, exposure 5 or more)
- **looks good**
FILE:templates/unit-review.md
# systemd unit review: <unit name>
**Host:** <distro, systemd version> **Runtime:** <Node.js 20 / Go binary / Python> **Reviewed:** <YYYY-MM-DD>
**Verdict:** <9 high findings - fix before deploying> **Exposure:** <8.1 before -> 1.5 after>
## Must fix
| # | Problem | Why it matters | Fix |
|---|---|---|---|
| 1 | <runs as root> | <an exploit owns the machine> | <DynamicUser=yes + CAP_NET_BIND_SERVICE for port 443> |
| 2 | <secret STRIPE_SECRET_KEY in Environment=> | <readable by every local user - rotate it> | <LoadCredential=stripe:/etc/shop-api/stripe> |
## Should fix
- <Restart/RestartSec, network-online pairing, [Install], stop timeout>
## Corrected unit
```ini
<full corrected .service file>
```
## Hardening drop-in (`systemctl edit <unit>`)
```ini
<drop-in from --dropin, adjusted for this service>
```
## Rollout
1. Rotate the leaked secrets; put the new ones in /etc/<name>/ (mode 0600).
2. `systemd-analyze verify /etc/systemd/system/<unit>`
3. `systemctl daemon-reload && systemctl restart <unit>`; check `journalctl -u <unit> -b` and the health endpoint.
4. `systemd-analyze security <unit>` - expect <exposure>.
5. Roll back: `systemctl revert <unit>` removes the drop-ins.
## Open questions
- [ ] <which directories does it write?>
- [ ] <does it need the network at start?>
FILE:examples/webapp.service
# /etc/systemd/system/shopapi.service - Node.js shop API, deployed by hand in 2023
[Unit]
Description=Shop API (Node.js)
After=network-online.target syslog.target
StartLimitBurst=3
[Service]
Type=simple
WorkingDirectory=/home/deploy/shop-api
Environment=NODE_ENV=production PORT=443
Environment="DATABASE_URL=postgres://shop:Winter2024@db.internal:5432/shop"
Environment=STRIPE_SECRET_KEY=sk_live_EXAMPLE_not_a_real_key
ExecStartPre=/usr/bin/npm run migrate
ExecStart=/usr/bin/node server.js >> /var/log/shop-api.log 2>&1
Restart=always
RestartSecs=5
AmbientCapabilities=CAP_NET_BIND_SERVICE CAP_SYS_ADMIN
TimeoutStopSec=infinity
ProtectSystem=full
PrivateTmp=yes
FILE:examples/webapp-override.conf
# /etc/systemd/system/shopapi.service.d/override.conf - added during an incident
[Service]
ProtectHome=yes
MemoryDenyWriteExecute=yes
Restart=on-faliure
FILE:examples/worker.service
# /etc/systemd/system/thumbnailer.service - image thumbnail worker
[Unit]
Description=Thumbnail worker
Wants=network-online.target
After=network-online.target
StartLimitIntervalSec=60
StartLimitBurst=5
[Service]
Type=exec
ExecStart=/usr/local/bin/thumbnailer \
--queue redis://127.0.0.1:6379/2 \
--out /var/lib/thumbnailer
DynamicUser=yes
StateDirectory=thumbnailer
LoadCredential=s3-key:/etc/thumbnailer/s3-key
Restart=on-failure
RestartSec=5s
TimeoutStopSec=30s
NoNewPrivileges=yes
ProtectSystem=strict
ProtectHome=yes
PrivateTmp=yes
PrivateDevices=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectControlGroups=yes
RestrictSUIDSGID=yes
RestrictNamespaces=yes
LockPersonality=yes
CapabilityBoundingSet=
SystemCallFilter=@system-service
SystemCallArchitectures=native
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
MemoryMax=1G
TasksMax=128
[Install]
WantedBy=multi-user.target
FILE:examples/review-report.md
# systemd unit review
| Unit | Description | Type | Runs as | Restart | Exposure (0-10) | High | Medium | Low | Info |
|---|---|---|---|---|---|---|---|---|---|
| `examples/webapp.service + examples/webapp-override.conf` | Shop API (Node.js) | simple | root | always | 8.1 UNSAFE | 9 | 4 | 2 | 1 |
| `examples/worker.service` | Thumbnail worker | exec | DynamicUser | on-failure | 1.2 OK | 0 | 0 | 1 | 0 |
## examples/webapp.service + examples/webapp-override.conf
**Verdict: 9 high finding(s) - fix before deploying**
| Sev | Code | Where | Finding | Fix |
|---|---|---|---|---|
| high | SD01 | examples/webapp.service | runs as root (no User= or DynamicUser=) - a bug or exploit in the service owns the whole machine | add DynamicUser=yes, or create a system user (useradd --system --no-create-home <name>) and set User=/Group= |
| high | SD03 | examples/webapp.service:11 | DATABASE_URL contains a URL with an embedded password | move the credential out of the unit (LoadCredential= / EnvironmentFile= 0600) |
| high | SD03 | examples/webapp.service:12 | secret STRIPE_SECRET_KEY is written in the unit file - unit files are world-readable and shown by `systemctl show` | use LoadCredential=stripe_secret_key:/etc/<name>/stripe_secret_key (read from $CREDENTIALS_DIRECTORY) or EnvironmentFile=/etc/<name>/env with mode 0600 |
| high | SD07 | examples/webapp.service:17 | grants CAP_SYS_ADMIN - close to full root | remove it; to bind ports below 1024 only CAP_NET_BIND_SERVICE is needed |
| high | SD08 | examples/webapp-override.conf:5 | invalid value Restart=on-faliure - systemd ignores the line and keeps the default (did you mean on-failure?) | use one of: always, no, on-abnormal, on-abort, on-failure, on-success, on-watchdog |
| high | SD08 | examples/webapp.service:16 | unknown directive RestartSecs= (ignored by systemd) - did you mean RestartSec=? | rename to RestartSec= |
| high | SD10 | examples/webapp.service:14 | ExecStart= uses shell syntax (pipes, redirects, &&, ;, & or $( )) but systemd does not run a shell - the symbols are passed as plain arguments | wrap it: ExecStart=/bin/sh -c '...' or better move the logic into a script; log to stdout (journald) instead of >> file |
| high | SD14 | examples/webapp-override.conf:3 | ProtectHome=yes but the service uses /home/deploy/shop-api - it cannot see it and will fail to start | move the app out of /home (e.g. /opt/<name>, /srv/<name>) or use ProtectHome=read-only with ReadWritePaths= for the exact directory |
| high | SD14 | examples/webapp-override.conf:4 | MemoryDenyWriteExecute=yes with a JIT runtime (/usr/bin/node) - it will crash on start | remove MemoryDenyWriteExecute= for this service |
| medium | SD05 | examples/webapp.service:4 | After=network-online.target without Wants=network-online.target - the ordering does nothing unless something else pulls the target in | add Wants=network-online.target |
| medium | SD06 | examples/webapp.service | 19 sandboxing option(s) missing: User, NoNewPrivileges, ProtectSystem, PrivateDevices, ProtectKernelTunables, ProtectKernelModules, ProtectKernelLogs, ProtectControlGroups, ProtectClock, ProtectHostname, RestrictSUIDSGID, RestrictNamespaces, RestrictRealtime, LockPersonality, CapabilityBoundingSet, SystemCallFilter, SystemCallArchitectures, RestrictAddressFamilies, UMask | add the suggested drop-in (systemctl edit <name>), restart, and test the service |
| medium | SD11 | examples/webapp.service | no [Install] WantedBy= - `systemctl enable` does nothing, so the service will not start at boot | add [Install] WantedBy=multi-user.target |
| medium | SD12 | examples/webapp.service:18 | stop timeout infinity - a hung service blocks shutdown and restarts forever | set TimeoutStopSec=30s (or what a clean shutdown really needs) |
| low | SD04 | examples/webapp.service:15 | Restart=always without RestartSec= (default 100ms) - a crash loop hammers the machine and hits the start limit fast | add RestartSec=5s |
| low | SD05 | examples/webapp.service:4 | After=syslog.target is obsolete | remove it (journald is always available; use network-online.target) |
| info | SD15 | examples/webapp.service | no resource limits (MemoryMax=, TasksMax=, CPUQuota=) - a leak can starve the host | add MemoryMax= and TasksMax= sized from normal usage |
Suggested drop-in (test after applying; relax an option if the service needs it):
```ini
# /etc/systemd/system/<name>.service.d/hardening.conf (systemctl edit <name>)
[Service]
# or User=<service account>; a non-root user needs AmbientCapabilities=CAP_NET_BIND_SERVICE for ports below 1024
DynamicUser=yes
NoNewPrivileges=yes
# currently full; add ReadWritePaths=/StateDirectory= for paths it writes
ProtectSystem=strict
PrivateDevices=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectKernelLogs=yes
ProtectControlGroups=yes
ProtectClock=yes
ProtectHostname=yes
RestrictSUIDSGID=yes
RestrictNamespaces=yes
RestrictRealtime=yes
LockPersonality=yes
# empty = no capabilities; or list only what you need, e.g. CAP_NET_BIND_SERVICE
CapabilityBoundingSet=
SystemCallFilter=@system-service
SystemCallArchitectures=native
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
UMask=0077
# then: systemctl daemon-reload && systemctl restart <name> && systemd-analyze security <name>
```
## examples/worker.service
**Verdict: looks good (exposure 1.2)**
| Sev | Code | Where | Finding | Fix |
|---|---|---|---|---|
| low | SD06 | examples/worker.service | 6 sandboxing option(s) missing: ProtectKernelLogs, ProtectClock, ProtectHostname, RestrictRealtime, MemoryDenyWriteExecute, UMask | add the suggested drop-in (systemctl edit <name>), restart, and test the service |
Suggested drop-in (test after applying; relax an option if the service needs it):
```ini
# /etc/systemd/system/<name>.service.d/hardening.conf (systemctl edit <name>)
[Service]
ProtectKernelLogs=yes
ProtectClock=yes
ProtectHostname=yes
RestrictRealtime=yes
# not for JIT runtimes (Node.js, Java, .NET, PyPy)
MemoryDenyWriteExecute=yes
UMask=0077
# then: systemctl daemon-reload && systemctl restart <name> && systemd-analyze security <name>
```
Exposure is this script's weighted estimate, not the systemd-analyze number; run `systemd-analyze security <unit>` and `systemd-analyze verify <file>` on the target host to confirm.
FILE:scripts/systemd_unit_review.py
#!/usr/bin/env python3
"""systemd service unit hardening reviewer (stdlib only, Python 3.8+).
Reads one or more systemd .service unit files (and drop-in .conf files, applied in the order
given) the way systemd does - sections, comments, backslash line continuation, repeated keys,
empty assignments that reset a list - and reviews them for reliability and security problems:
running as root, secrets in Environment=, shell syntax in Exec lines without a shell, unknown or
misspelled directives and values, directives in the wrong section, restart and timeout policy,
network-online ordering, missing [Install], sandboxing gaps, dangerous capabilities and options
that will stop the service from starting. Prints an exposure score (0 = locked down, 10 = wide
open, in the spirit of `systemd-analyze security`) and a suggested hardening drop-in.
Usage:
systemd_unit_review.py UNIT.service [DROPIN.conf ...] [--json] [--dropin]
systemd_unit_review.py --self-test
Exit: 0 = no high findings, 1 = at least one high finding, 2 = bad input.
"""
import argparse, difflib, json, re, sys
SEV = {"high": 0, "medium": 1, "low": 2, "info": 3}
UNIT_KEYS = {"Description", "Documentation", "Requires", "Requisite", "Wants", "BindsTo", "PartOf",
"Upholds", "Conflicts", "Before", "After", "OnFailure", "OnSuccess", "PropagatesReloadTo",
"ReloadPropagatedFrom", "JoinsNamespaceOf", "RequiresMountsFor", "OnFailureJobMode",
"IgnoreOnIsolate", "StopWhenUnneeded", "RefuseManualStart", "RefuseManualStop",
"AllowIsolate", "DefaultDependencies", "CollectMode", "FailureAction", "SuccessAction",
"JobTimeoutSec", "JobRunningTimeoutSec", "StartLimitIntervalSec", "StartLimitBurst",
"StartLimitAction", "RebootArgument", "SourcePath", "ConditionPathExists",
"ConditionPathIsDirectory", "ConditionFileNotEmpty", "ConditionHost", "ConditionVirtualization",
"ConditionKernelCommandLine", "ConditionACPower", "ConditionUser", "ConditionGroup",
"AssertPathExists", "AssertFileNotEmpty", "ConditionEnvironment", "ConditionSecurity"}
SERVICE_KEYS = {"Type", "ExitType", "RemainAfterExit", "GuessMainPID", "PIDFile", "BusName",
"ExecStart", "ExecStartPre", "ExecStartPost", "ExecCondition", "ExecReload", "ExecStop",
"ExecStopPost", "RestartSec", "RestartSteps", "RestartMaxDelaySec", "TimeoutStartSec",
"TimeoutStopSec", "TimeoutAbortSec", "TimeoutSec", "TimeoutStartFailureMode",
"TimeoutStopFailureMode", "RuntimeMaxSec", "WatchdogSec", "Restart", "RestartMode",
"SuccessExitStatus", "RestartPreventExitStatus", "RestartForceExitStatus", "RootDirectoryStartOnly",
"NonBlocking", "NotifyAccess", "Sockets", "FileDescriptorStoreMax", "USBFunctionDescriptors",
"OOMPolicy", "OpenFile", "ReloadSignal",
# exec
"WorkingDirectory", "RootDirectory", "RootImage", "User", "Group", "DynamicUser",
"SupplementaryGroups", "PAMName", "UMask", "Environment", "EnvironmentFile", "PassEnvironment",
"UnsetEnvironment", "StandardInput", "StandardOutput", "StandardError", "SyslogIdentifier",
"LogLevelMax", "LogRateLimitIntervalSec", "LogRateLimitBurst", "Nice", "OOMScoreAdjust",
"CPUSchedulingPolicy", "IOSchedulingClass", "LimitNOFILE", "LimitNPROC", "LimitCORE",
"LimitMEMLOCK", "LimitAS", "LimitFSIZE", "CapabilityBoundingSet", "AmbientCapabilities",
"NoNewPrivileges", "SecureBits", "ProtectSystem", "ProtectHome", "RuntimeDirectory",
"StateDirectory", "CacheDirectory", "LogsDirectory", "ConfigurationDirectory",
"RuntimeDirectoryMode", "StateDirectoryMode", "RuntimeDirectoryPreserve", "ReadWritePaths",
"ReadOnlyPaths", "InaccessiblePaths", "ExecPaths", "NoExecPaths", "TemporaryFileSystem",
"BindPaths", "BindReadOnlyPaths", "PrivateTmp", "PrivateDevices", "PrivateNetwork",
"PrivateUsers", "PrivateIPC", "PrivateMounts", "ProtectHostname", "ProtectClock",
"ProtectKernelTunables", "ProtectKernelModules", "ProtectKernelLogs", "ProtectControlGroups",
"ProtectProc", "ProcSubset", "RestrictAddressFamilies", "RestrictNamespaces",
"LockPersonality", "MemoryDenyWriteExecute", "RestrictRealtime", "RestrictSUIDSGID",
"RemoveIPC", "SystemCallFilter", "SystemCallArchitectures", "SystemCallErrorNumber",
"DevicePolicy", "DeviceAllow", "IPAddressAllow", "IPAddressDeny", "LoadCredential",
"LoadCredentialEncrypted", "SetCredential", "SetCredentialEncrypted", "KeyringMode",
"TTYPath", "UtmpIdentifier", "SELinuxContext", "AppArmorProfile", "KillMode", "KillSignal",
"SendSIGKILL", "SendSIGHUP", "FinalKillSignal", "RestartKillSignal",
# resource control
"CPUQuota", "CPUWeight", "MemoryMax", "MemoryHigh", "MemoryLow", "MemoryMin", "MemorySwapMax",
"TasksMax", "IOWeight", "Slice", "Delegate", "CPUAccounting", "MemoryAccounting",
"IOAccounting", "TasksAccounting", "IPAccounting", "ManagedOOMSwap",
"ManagedOOMMemoryPressure"}
INSTALL_KEYS = {"Alias", "WantedBy", "RequiredBy", "UpheldBy", "Also", "DefaultInstance"}
SECTIONS = {"Unit": UNIT_KEYS, "Service": SERVICE_KEYS, "Install": INSTALL_KEYS}
LIST_KEYS = {"Requires", "Wants", "After", "Before", "BindsTo", "PartOf", "Conflicts", "Documentation",
"Environment", "EnvironmentFile", "ExecStartPre", "ExecStartPost", "ExecStop", "ExecStopPost",
"ExecReload", "ExecCondition", "ReadWritePaths", "ReadOnlyPaths", "InaccessiblePaths",
"SupplementaryGroups", "CapabilityBoundingSet", "AmbientCapabilities", "SystemCallFilter",
"RestrictAddressFamilies", "DeviceAllow", "LoadCredential", "SetCredential", "WantedBy",
"RequiredBy", "Alias", "Also", "BindPaths", "BindReadOnlyPaths", "IPAddressAllow",
"IPAddressDeny", "OnFailure", "SuccessExitStatus", "RestartPreventExitStatus",
"TemporaryFileSystem", "PassEnvironment", "UnsetEnvironment", "RequiresMountsFor",
"StateDirectory", "RuntimeDirectory", "CacheDirectory", "LogsDirectory", "ExecStart",
"ConditionPathExists", "AssertPathExists"}
VALUES = {
"Type": {"simple", "exec", "forking", "oneshot", "dbus", "notify", "notify-reload", "idle"},
"Restart": {"no", "on-success", "on-failure", "on-abnormal", "on-watchdog", "on-abort", "always"},
"ProtectSystem": {"yes", "no", "true", "false", "full", "strict", "1", "0", "on", "off"},
"ProtectHome": {"yes", "no", "true", "false", "read-only", "tmpfs", "1", "0", "on", "off"},
"KillMode": {"control-group", "mixed", "process", "none"},
"StandardOutput": None, "NotifyAccess": {"none", "main", "exec", "all"},
}
BOOL = {"yes", "no", "true", "false", "1", "0", "on", "off"}
BOOL_KEYS = {"NoNewPrivileges", "PrivateTmp", "PrivateDevices", "PrivateNetwork", "PrivateUsers",
"ProtectKernelTunables", "ProtectKernelModules", "ProtectKernelLogs", "ProtectControlGroups",
"ProtectHostname", "ProtectClock", "LockPersonality", "MemoryDenyWriteExecute",
"RestrictRealtime", "RestrictSUIDSGID", "RemoveIPC", "DynamicUser", "RemainAfterExit",
"PrivateIPC", "PrivateMounts", "DefaultDependencies"}
TRUE = {"yes", "true", "1", "on"}
SECRET_RE = re.compile(r"(PASS(WORD|WD)?|SECRET|TOKEN|API_?KEY|PRIVATE_?KEY|CREDENTIALS?|DSN)", re.I)
SHELL_RE = re.compile(r"(\|\||&&|(?<![\\])[|;]|>>?|<|\$\(|`|(?:^|\s)&\s*$|\s&$)")
DANGER_CAPS = {"CAP_SYS_ADMIN": "high", "CAP_SYS_MODULE": "high", "CAP_SYS_PTRACE": "high",
"CAP_DAC_OVERRIDE": "medium", "CAP_SYS_RAWIO": "high", "CAP_NET_ADMIN": "medium",
"CAP_SETUID": "medium", "CAP_SETGID": "medium", "CAP_CHOWN": "low", "CAP_SYS_CHROOT": "low"}
# (directive, wanted value test, weight, recommended line) - weights sum to 10.0
HARDEN = [
("User", None, 2.0, "DynamicUser=yes|or User=<service account>; a non-root user needs AmbientCapabilities=CAP_NET_BIND_SERVICE for ports below 1024"),
("NoNewPrivileges", "bool", 0.8, "NoNewPrivileges=yes"),
("ProtectSystem", "strict", 1.0, "ProtectSystem=strict"),
("ProtectHome", "home", 0.6, "ProtectHome=yes"),
("PrivateTmp", "bool", 0.5, "PrivateTmp=yes"),
("PrivateDevices", "bool", 0.5, "PrivateDevices=yes"),
("ProtectKernelTunables", "bool", 0.4, "ProtectKernelTunables=yes"),
("ProtectKernelModules", "bool", 0.4, "ProtectKernelModules=yes"),
("ProtectKernelLogs", "bool", 0.2, "ProtectKernelLogs=yes"),
("ProtectControlGroups", "bool", 0.3, "ProtectControlGroups=yes"),
("ProtectClock", "bool", 0.2, "ProtectClock=yes"),
("ProtectHostname", "bool", 0.1, "ProtectHostname=yes"),
("RestrictSUIDSGID", "bool", 0.3, "RestrictSUIDSGID=yes"),
("RestrictNamespaces", "ns", 0.4, "RestrictNamespaces=yes"),
("RestrictRealtime", "bool", 0.2, "RestrictRealtime=yes"),
("LockPersonality", "bool", 0.1, "LockPersonality=yes"),
("MemoryDenyWriteExecute", "bool", 0.3, "MemoryDenyWriteExecute=yes|not for JIT runtimes (Node.js, Java, .NET, PyPy)"),
("CapabilityBoundingSet", "caps", 0.6, "CapabilityBoundingSet=|empty = no capabilities; or list only what you need, e.g. CAP_NET_BIND_SERVICE"),
("SystemCallFilter", "set", 0.4, "SystemCallFilter=@system-service"),
("SystemCallArchitectures", "native", 0.1, "SystemCallArchitectures=native"),
("RestrictAddressFamilies", "set", 0.4, "RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6"),
("UMask", "umask", 0.2, "UMask=0077"),
]
class InputError(Exception):
pass
# ---------------------------------------------------------------- parse
def parse(text, fname, unit=None):
"""Return unit dict: {section: {key: [(value, line, file)]}} plus raw entries list."""
unit = unit if unit is not None else {"sections": {}, "entries": [], "files": []}
unit["files"].append(fname)
sec = None
lines = text.splitlines()
i = 0
while i < len(lines):
ln, raw = i + 1, lines[i]
i += 1
s = raw.strip()
if not s or s[0] in "#;":
continue
while s.endswith("\\") and i < len(lines):
nxt = lines[i].strip()
i += 1
if nxt.startswith(("#", ";")):
continue
s = s[:-1].rstrip() + " " + nxt
if s.startswith("["):
if not s.endswith("]"):
raise InputError("%s:%d: broken section header %r" % (fname, ln, s))
sec = s[1:-1].strip()
unit["sections"].setdefault(sec, {})
unit["entries"].append(("section", sec, None, ln, fname))
continue
if "=" not in s:
unit["entries"].append(("noeq", sec, s, ln, fname))
continue
if sec is None:
unit["entries"].append(("nosec", None, s, ln, fname))
continue
k, v = s.split("=", 1)
k, v = k.strip(), v.strip()
unit["entries"].append(("kv", sec, (k, v), ln, fname))
d = unit["sections"][sec]
if v == "" and k in LIST_KEYS:
d[k] = [("", ln, fname)] # empty assignment resets the list
continue
d.setdefault(k, []).append((v, ln, fname))
return unit
def get(unit, sec, key):
vals = unit["sections"].get(sec, {}).get(key, [])
vals = [v for v in vals if v[0] != ""] if key in LIST_KEYS else vals
return vals
def valid(key, v):
vl = v.lower()
if key in VALUES and VALUES[key]:
return vl in VALUES[key]
if key in BOOL_KEYS:
return vl in BOOL
return True
def last(unit, sec, key, default=None):
"""Effective value: the last assignment systemd accepts (invalid values are ignored)."""
for v in reversed(unit["sections"].get(sec, {}).get(key) or []):
if v[0] == "" or valid(key, v[0]):
return v[0]
return default
def where(unit, sec, key):
v = [x for x in unit["sections"].get(sec, {}).get(key) or [] if x[0] == "" or valid(key, x[0])]
return "%s:%d" % (v[-1][2], v[-1][1]) if v else unit["files"][0]
def words(unit, sec, key):
out = []
for v, _, _ in get(unit, sec, key):
out += v.split()
return out
# ---------------------------------------------------------------- review
def review(unit):
F = []
add = lambda sev, code, loc, msg, fix: F.append((sev, code, loc, msg, fix))
S = unit["sections"]
svc = S.get("Service", {})
f0 = unit["files"][0]
if "Service" not in S:
raise InputError("%s has no [Service] section - not a service unit" % f0)
# SD08 / SD13 / structure
for kind, sec, val, ln, fn in unit["entries"]:
loc = "%s:%d" % (fn, ln)
if kind == "noeq":
add("medium", "SD08", loc, "line is not Key=Value and is ignored: %r" % val, "fix the line or comment it out with #")
elif kind == "nosec":
add("medium", "SD08", loc, "assignment before any [Section] is ignored: %r" % val, "move it under [Unit], [Service] or [Install]")
elif kind == "section" and sec not in SECTIONS and not sec.startswith("X-"):
add("medium", "SD08", loc, "unknown section [%s] - its settings are ignored" % sec, "use [Unit], [Service] or [Install] (custom sections must start with X-)")
elif kind == "kv" and sec in SECTIONS:
k, v = val
if k.startswith("X-"):
continue
if k not in SECTIONS[sec]:
other = [s for s in SECTIONS if k in SECTIONS[s]]
if other:
sev = "medium"
add(sev, "SD09", loc, "%s= belongs in [%s], not [%s] - systemd ignores it here" % (k, other[0], sec),
"move %s=%s to [%s]" % (k, v, other[0]))
else:
allk = set().union(*SECTIONS.values())
close = difflib.get_close_matches(k, sorted(allk), 1, 0.75)
add("high" if sec == "Service" and close and close[0].startswith(("Exec", "User", "Restart")) else "medium",
"SD08", loc, "unknown directive %s= (ignored by systemd)%s" % (k, " - did you mean %s=?" % close[0] if close else ""),
"rename to %s=" % close[0] if close else "remove it or check `man systemd.directives`")
continue
vl = v.lower()
if k in VALUES and VALUES[k] and v and vl not in VALUES[k]:
close = difflib.get_close_matches(vl, sorted(VALUES[k]), 1, 0.6)
add("high" if k in ("Type", "Restart") else "medium", "SD08", loc,
"invalid value %s=%s - systemd ignores the line and keeps the default%s" % (k, v, " (did you mean %s?)" % close[0] if close else ""),
"use one of: %s" % ", ".join(sorted(VALUES[k])))
elif k in BOOL_KEYS and v and vl not in BOOL:
add("medium", "SD08", loc, "%s= expects a boolean, got %r (line ignored)" % (k, v), "use yes or no")
seen = {}
for kind, sec, val, ln, fn in unit["entries"]:
if kind == "kv" and val[0] not in LIST_KEYS and sec in SECTIONS:
key = (sec, val[0])
if key in seen and seen[key][2] == fn:
add("low", "SD13", "%s:%d" % (fn, ln), "%s= set again (first at line %d) - the last one wins: %s" % (val[0], seen[key][1], val[1]),
"keep one %s= line" % val[0])
seen[key] = (val[1], ln, fn)
typ = (last(unit, "Service", "Type", "simple") or "simple").lower()
execs = get(unit, "Service", "ExecStart")
# SD02 ExecStart
if not execs:
add("high", "SD02", f0, "no ExecStart= - the service cannot start", "add ExecStart=/absolute/path/to/binary args")
if len(execs) > 1 and typ != "oneshot":
add("high", "SD02", where(unit, "Service", "ExecStart"), "%d ExecStart= lines but Type=%s allows only one (only oneshot may have several)" % (len(execs), typ),
"keep one ExecStart=, or reset with an empty ExecStart= first in a drop-in")
for key in ("ExecStartPre", "ExecStart", "ExecStartPost", "ExecReload", "ExecStop", "ExecStopPost"):
for v, ln, fn in get(unit, "Service", key):
loc = "%s:%d" % (fn, ln)
cmd = v.lstrip("-@:+!")
first = cmd.split()[0] if cmd.split() else ""
is_shell = re.match(r"^(/usr)?/bin/(ba|da|z)?sh$", first) is not None
if first and not first.startswith("/") and not first.startswith("$"):
add("low", "SD02", loc, "%s= command %r is not an absolute path (systemd searches only its fixed PATH)" % (key, first),
"use the full path, e.g. /usr/bin/%s" % first)
if not is_shell and SHELL_RE.search(cmd.split(None, 1)[1] if len(cmd.split(None, 1)) > 1 else ""):
add("high", "SD10", loc, "%s= uses shell syntax (pipes, redirects, &&, ;, & or $( )) but systemd does not run a shell - the symbols are passed as plain arguments" % key,
"wrap it: %s=/bin/sh -c '...' or better move the logic into a script; log to stdout (journald) instead of >> file" % key)
if re.search(r"(^|\s)(nohup|daemon|--daemon(ize)?|-d)(\s|$)", cmd) and typ in ("simple", "exec"):
add("medium", "SD10", loc, "%s= looks like it forks into the background (nohup/--daemon) while Type=%s expects the process to stay in the foreground" % (key, typ),
"run it in the foreground (drop --daemon / nohup) or use Type=forking with PIDFile=")
if v.startswith("+") or v.startswith("!"):
add("medium", "SD01", loc, "%s= prefix %s runs this command with full privileges, bypassing User= and the sandbox" % (key, v[0]), "drop the prefix unless this step really needs root")
if typ == "forking" and not get(unit, "Service", "PIDFile"):
add("low", "SD10", where(unit, "Service", "Type"), "Type=forking without PIDFile= - systemd has to guess the main process", "add PIDFile=/run/<name>/<name>.pid or run in the foreground with Type=exec")
# SD01 user
user = last(unit, "Service", "User")
dyn = (last(unit, "Service", "DynamicUser", "no") or "no").lower() in TRUE
root = (not dyn) and (user is None or user in ("root", "0"))
if root:
add("high" if typ != "oneshot" else "medium", "SD01", where(unit, "Service", "User") if user else f0,
"runs as root (%s) - a bug or exploit in the service owns the whole machine" % ("User=%s" % user if user else "no User= or DynamicUser="),
"add DynamicUser=yes, or create a system user (useradd --system --no-create-home <name>) and set User=/Group=")
caps = words(unit, "Service", "AmbientCapabilities") + words(unit, "Service", "CapabilityBoundingSet")
for c in sorted(set(x.lstrip("~") for x in caps)):
if c.upper() in DANGER_CAPS and not any(x.startswith("~") for x in caps):
add(DANGER_CAPS[c.upper()], "SD07", where(unit, "Service", "AmbientCapabilities" if c in words(unit, "Service", "AmbientCapabilities") else "CapabilityBoundingSet"),
"grants %s - close to full root" % c if DANGER_CAPS[c.upper()] == "high" else "grants %s" % c,
"remove it; to bind ports below 1024 only CAP_NET_BIND_SERVICE is needed")
# SD03 secrets
for v, ln, fn in get(unit, "Service", "Environment"):
for m in re.finditer(r'"?([A-Za-z_][A-Za-z0-9_]*)=("[^"]*"|\S*)', v):
name, val = m.group(1), m.group(2).strip('"')
if SECRET_RE.search(name) and val and not val.startswith("%") and not val.startswith("$"):
add("high", "SD03", "%s:%d" % (fn, ln), "secret %s is written in the unit file - unit files are world-readable and shown by `systemctl show`" % name,
"use LoadCredential=%s:/etc/<name>/%s (read from $CREDENTIALS_DIRECTORY) or EnvironmentFile=/etc/<name>/env with mode 0600" % (name.lower(), name.lower()))
if re.search(r"://[^/\s:@]+:[^@\s]+@", val):
add("high", "SD03", "%s:%d" % (fn, ln), "%s contains a URL with an embedded password" % name, "move the credential out of the unit (LoadCredential= / EnvironmentFile= 0600)")
for v, ln, fn in get(unit, "Service", "ExecStart"):
if re.search(r"--(password|passwd|token|secret|api-key)[= ]\S+", v, re.I):
add("high", "SD03", "%s:%d" % (fn, ln), "a secret is passed on the command line - visible to every user in ps and /proc", "read it from a file or $CREDENTIALS_DIRECTORY instead")
# SD04 restart policy
restart = (last(unit, "Service", "Restart", "no") or "no").lower()
if typ != "oneshot" and restart == "no":
add("medium", "SD04", where(unit, "Service", "Restart") if last(unit, "Service", "Restart") else f0,
"Restart=no (default) - a crash leaves the service down until someone notices", "add Restart=on-failure and RestartSec=5s")
if restart in ("always", "on-failure", "on-abnormal") and not last(unit, "Service", "RestartSec"):
add("low", "SD04", where(unit, "Service", "Restart"), "Restart=%s without RestartSec= (default 100ms) - a crash loop hammers the machine and hits the start limit fast" % restart, "add RestartSec=5s")
if typ == "oneshot" and restart in ("always", "on-success"):
add("high", "SD04", where(unit, "Service", "Restart"), "Type=oneshot does not allow Restart=%s - systemd refuses to load the unit" % restart, "use Restart=on-failure or run it from a .timer")
tstop = (last(unit, "Service", "TimeoutStopSec") or last(unit, "Service", "TimeoutSec") or "").lower()
if tstop in ("infinity", "0"):
add("medium", "SD12", where(unit, "Service", "TimeoutStopSec") if last(unit, "Service", "TimeoutStopSec") else where(unit, "Service", "TimeoutSec"),
"stop timeout %s - a hung service blocks shutdown and restarts forever" % tstop, "set TimeoutStopSec=30s (or what a clean shutdown really needs)")
km = (last(unit, "Service", "KillMode", "") or "").lower()
if km == "none":
add("medium", "SD12", where(unit, "Service", "KillMode"), "KillMode=none is deprecated - leftover processes survive stop and restart", "use KillMode=mixed or the default control-group")
elif km == "process":
add("low", "SD12", where(unit, "Service", "KillMode"), "KillMode=process kills only the main process; children keep running after stop", "use KillMode=mixed unless the children must outlive the service")
# SD05 ordering
after = words(unit, "Unit", "After")
wants = words(unit, "Unit", "Wants") + words(unit, "Unit", "Requires")
if "network-online.target" in after and "network-online.target" not in wants:
add("medium", "SD05", where(unit, "Unit", "After"), "After=network-online.target without Wants=network-online.target - the ordering does nothing unless something else pulls the target in",
"add Wants=network-online.target")
if "network-online.target" in wants and "network-online.target" not in after:
add("medium", "SD05", where(unit, "Unit", "Wants"), "Wants=network-online.target without After=network-online.target - the service still starts before the network is up", "add After=network-online.target")
for v in ("syslog.target", "network.service"):
if v in after:
add("low", "SD05", where(unit, "Unit", "After"), "After=%s is obsolete" % v, "remove it (journald is always available; use network-online.target)")
sl = [k for k in ("StartLimitIntervalSec", "StartLimitBurst") if k in S.get("Unit", {})]
for k, ln in (("StartLimitInterval", None),):
if k in svc:
add("medium", "SD09", where(unit, "Service", k), "StartLimitInterval= is the old name and in [Service] it is ignored", "use StartLimitIntervalSec= in [Unit]")
if restart == "always" and not sl and "StartLimitIntervalSec" not in svc:
add("info", "SD04", f0, "Restart=always with the default start limit (5 starts in 10 s) - after that the unit stays failed", "if you want endless retries set StartLimitIntervalSec=0 in [Unit] and RestartSec= to a sane delay")
# SD11 install
wb = words(unit, "Install", "WantedBy") + words(unit, "Install", "RequiredBy")
if not wb and not get(unit, "Install", "Alias"):
add("medium", "SD11", f0, "no [Install] WantedBy= - `systemctl enable` does nothing, so the service will not start at boot", "add [Install] WantedBy=multi-user.target")
elif "default.target" in wb or "graphical.target" in wb:
add("low", "SD11", where(unit, "Install", "WantedBy"), "WantedBy=%s for a system service" % [w for w in wb if w in ("default.target", "graphical.target")][0], "use WantedBy=multi-user.target (default.target is for user services)")
# SD14 conflicts that break start
ph = (last(unit, "Service", "ProtectHome", "no") or "no").lower()
wd = last(unit, "Service", "WorkingDirectory", "") or ""
paths = [wd] + [(e[0].lstrip("-@:+!").split() or [""])[0] for e in execs] + [x.lstrip("-") for x in words(unit, "Service", "EnvironmentFile")]
if ph in TRUE or ph in ("tmpfs", "read-only"):
for p in paths:
if p.startswith("/home/") or p.startswith("/root/"):
add("high" if ph != "read-only" else "medium", "SD14", where(unit, "Service", "ProtectHome"),
"ProtectHome=%s but the service uses %s - it %s" % (ph, p, "cannot see it and will fail to start" if ph != "read-only" else "cannot write there"),
"move the app out of /home (e.g. /opt/<name>, /srv/<name>) or use ProtectHome=read-only with ReadWritePaths= for the exact directory")
break
ps_ = (last(unit, "Service", "ProtectSystem", "no") or "no").lower()
rw = words(unit, "Service", "ReadWritePaths") + ["/var/lib/" + x for x in words(unit, "Service", "StateDirectory")] + \
["/var/log/" + x for x in words(unit, "Service", "LogsDirectory")] + ["/run/" + x for x in words(unit, "Service", "RuntimeDirectory")]
if ps_ == "strict":
for v, ln, fn in execs + get(unit, "Service", "ExecStartPre"):
for m in re.finditer(r">>?\s*(/\S+)|--(?:log|pid|data|out)[-a-z]*[= ](/\S+)", v):
p = m.group(1) or m.group(2)
if not any(p.startswith(x.lstrip("-")) for x in rw) and not p.startswith(("/tmp", "/dev/")):
add("medium", "SD14", "%s:%d" % (fn, ln), "ProtectSystem=strict makes %s read-only - writes will fail" % p,
"use StateDirectory=/LogsDirectory=/RuntimeDirectory= or ReadWritePaths=%s" % p.rsplit("/", 1)[0])
if (last(unit, "Service", "MemoryDenyWriteExecute", "no") or "").lower() in TRUE and paths and re.search(r"(node|java|dotnet|pypy|bun|deno)", " ".join(paths)):
add("high", "SD14", where(unit, "Service", "MemoryDenyWriteExecute"), "MemoryDenyWriteExecute=yes with a JIT runtime (%s) - it will crash on start" % [p for p in paths if re.search(r"(node|java|dotnet|pypy|bun|deno)", p)][0],
"remove MemoryDenyWriteExecute= for this service")
if dyn and user:
add("info", "SD01", where(unit, "Service", "DynamicUser"), "DynamicUser=yes with User=%s - a transient user with that name is allocated" % user, "fine; do not rely on files owned by a static %s user" % user)
# SD15 resources
if not any(k in svc for k in ("MemoryMax", "MemoryHigh", "TasksMax", "CPUQuota")):
add("info", "SD15", f0, "no resource limits (MemoryMax=, TasksMax=, CPUQuota=) - a leak can starve the host", "add MemoryMax= and TasksMax= sized from normal usage")
# SD06 exposure
exposure, missing = 0.0, []
for key, test, w, rec in HARDEN:
v = (last(unit, "Service", key) or "")
vl = v.lower()
if key == "User":
ok = not root
elif test == "bool":
ok = vl in TRUE
elif test == "strict":
ok = vl == "strict"
if vl in ("full", "yes", "true"):
exposure += w * 0.5
missing.append((key, "ProtectSystem=strict|currently %s; add ReadWritePaths=/StateDirectory= for paths it writes" % v))
continue
elif test == "home":
ok = vl in TRUE or vl in ("read-only", "tmpfs")
elif test == "ns":
ok = vl in TRUE or (v != "" and vl not in ("no", "false", "0", "off"))
elif test == "caps":
ok = key in svc and all(c.upper() not in DANGER_CAPS or c.startswith("~") for c in words(unit, "Service", key))
elif test == "set":
ok = bool(words(unit, "Service", key))
elif test == "native":
ok = vl == "native"
elif test == "umask":
ok = v in ("0077", "077", "0027", "027")
else:
ok = bool(v)
if not ok:
exposure += w
missing.append((key, rec))
exposure = round(min(exposure, 10.0), 1)
if missing:
add("medium" if exposure >= 5 else "low", "SD06", f0, "%d sandboxing option(s) missing: %s" % (len(missing), ", ".join(k for k, _ in missing)),
"add the suggested drop-in (systemctl edit <name>), restart, and test the service")
F.sort(key=lambda x: (SEV[x[0]], x[1], x[2]))
rating = "OK" if exposure < 2 else "MEDIUM" if exposure < 5 else "EXPOSED" if exposure < 8 else "UNSAFE"
return {"files": unit["files"], "name": last(unit, "Unit", "Description", "") or unit["files"][0], "type": typ,
"user": "DynamicUser" if dyn else (user or "root"), "restart": restart, "exposure": exposure, "rating": rating,
"missing": missing, "findings": F}
def dropin(r):
lines = ["# /etc/systemd/system/<name>.service.d/hardening.conf (systemctl edit <name>)", "[Service]"]
for k, rec in r["missing"]:
line, _, note = rec.partition("|")
if note:
lines.append("# " + note)
lines.append(line)
if any(f[1] == "SD04" and "Restart=no" in f[3] for f in r["findings"]):
lines += ["Restart=on-failure", "RestartSec=5s"]
lines += ["# then: systemctl daemon-reload && systemctl restart <name> && systemd-analyze security <name>"]
return "\n".join(lines)
def counts(F):
c = {s: 0 for s in SEV}
for f in F:
c[f[0]] += 1
return c
def report(results):
out = ["# systemd unit review", ""]
out.append("| Unit | Description | Type | Runs as | Restart | Exposure (0-10) | High | Medium | Low | Info |")
out.append("|---|---|---|---|---|---|---|---|---|---|")
for r in results:
c = counts(r["findings"])
out.append("| `%s` | %s | %s | %s | %s | %.1f %s | %d | %d | %d | %d |" % (" + ".join(r["files"]), r["name"], r["type"], r["user"], r["restart"],
r["exposure"], r["rating"], c["high"], c["medium"], c["low"], c["info"]))
for r in results:
c = counts(r["findings"])
out += ["", "## %s" % " + ".join(r["files"]), ""]
if c["high"]:
v = "%d high finding(s) - fix before deploying" % c["high"]
elif r["exposure"] >= 5:
v = "works, but exposure %.1f - add the hardening drop-in" % r["exposure"]
else:
v = "looks good (exposure %.1f)" % r["exposure"]
out.append("**Verdict: %s**" % v)
out += ["", "| Sev | Code | Where | Finding | Fix |", "|---|---|---|---|---|"]
for s, code, loc, msg, fix in r["findings"]:
out.append("| %s | %s | %s | %s | %s |" % (s, code, loc, msg.replace("|", "\\|"), fix.replace("|", "\\|")))
if r["missing"]:
out += ["", "Suggested drop-in (test after applying; relax an option if the service needs it):", "", "```ini", dropin(r), "```"]
out += ["", "Exposure is this script's weighted estimate, not the systemd-analyze number; run `systemd-analyze security <unit>` and `systemd-analyze verify <file>` on the target host to confirm."]
return "\n".join(out) + "\n"
def run_files(paths):
unit = None
for p in paths:
try:
with open(p, encoding="utf-8") as fh:
txt = fh.read()
except OSError as e:
raise InputError("cannot read %s: %s" % (p, e))
unit = parse(txt, p, unit)
return review(unit)
# ---------------------------------------------------------------- self-test
def self_test():
tests = []
def t(name, cond):
tests.append((name, bool(cond)))
def R(txt, *drop):
u = parse(txt, "a.service")
for i, d in enumerate(drop):
u = parse(d, "d%d.conf" % i, u)
return review(u)
def codes(txt, *drop):
return [(f[0], f[1]) for f in R(txt, *drop)["findings"]]
base = "[Unit]\nDescription=x\n[Service]\nExecStart=/usr/bin/app\n[Install]\nWantedBy=multi-user.target\n"
good = ("[Unit]\nDescription=good\nWants=network-online.target\nAfter=network-online.target\n[Service]\nType=exec\nExecStart=/usr/bin/app\n"
"DynamicUser=yes\nRestart=on-failure\nRestartSec=5s\nNoNewPrivileges=yes\nProtectSystem=strict\nProtectHome=yes\nPrivateTmp=yes\n"
"PrivateDevices=yes\nProtectKernelTunables=yes\nProtectKernelModules=yes\nProtectKernelLogs=yes\nProtectControlGroups=yes\n"
"ProtectClock=yes\nProtectHostname=yes\nRestrictSUIDSGID=yes\nRestrictNamespaces=yes\nRestrictRealtime=yes\nLockPersonality=yes\n"
"MemoryDenyWriteExecute=yes\nCapabilityBoundingSet=\nSystemCallFilter=@system-service\nSystemCallArchitectures=native\n"
"RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6\nUMask=0077\nMemoryMax=512M\nTasksMax=64\n[Install]\nWantedBy=multi-user.target\n")
g = R(good)
t("fully hardened unit: exposure 0, no high/medium", g["exposure"] == 0 and not [f for f in g["findings"] if f[0] in ("high", "medium")])
t("root by default -> SD01 high", ("high", "SD01") in codes(base))
t("root oneshot -> SD01 medium", ("medium", "SD01") in codes(base.replace("[Service]\n", "[Service]\nType=oneshot\n")))
t("User=svc clears SD01", "SD01" not in [c for _, c in codes(base.replace("[Service]\n", "[Service]\nUser=svc\n"))])
t("missing ExecStart -> SD02 high", ("high", "SD02") in codes("[Service]\nUser=a\n[Install]\nWantedBy=multi-user.target\n"))
t("relative command -> SD02 low", ("low", "SD02") in codes(base.replace("/usr/bin/app", "app")))
t("two ExecStart with simple -> SD02 high", ("high", "SD02") in codes(base.replace("ExecStart=/usr/bin/app", "ExecStart=/a\nExecStart=/b")))
t("empty ExecStart= in drop-in resets", ("high", "SD02") not in codes(base, "[Service]\nExecStart=\nExecStart=/usr/bin/other\n"))
t("pipe in ExecStart -> SD10 high", ("high", "SD10") in codes(base.replace("/usr/bin/app", "/usr/bin/app | logger")))
t("redirect >> -> SD10 high", ("high", "SD10") in codes(base.replace("/usr/bin/app", "/usr/bin/app >> /var/log/app.log")))
t("sh -c is allowed", ("high", "SD10") not in codes(base.replace("/usr/bin/app", "/bin/sh -c '/usr/bin/app | logger'")))
t("--daemon with simple -> SD10 medium", ("medium", "SD10") in codes(base.replace("/usr/bin/app", "/usr/bin/app --daemon")))
t("secret in Environment -> SD03 high", ("high", "SD03") in codes(base.replace("[Service]\n", "[Service]\nEnvironment=DB_PASSWORD=hunter2\n")))
t("quoted env with URL password -> SD03", ("high", "SD03") in codes(base.replace("[Service]\n", '[Service]\nEnvironment="DATABASE_URL=postgres://u:pw@db/x" "PORT=80"\n')))
t("non-secret env is fine", "SD03" not in [c for _, c in codes(base.replace("[Service]\n", "[Service]\nEnvironment=PORT=8080 LOG_LEVEL=info\n"))])
t("--password on cmdline -> SD03", ("high", "SD03") in codes(base.replace("/usr/bin/app", "/usr/bin/app --password=s3cret")))
t("Restart missing -> SD04 medium", ("medium", "SD04") in codes(base))
t("Restart=always no RestartSec -> SD04 low", ("low", "SD04") in codes(base.replace("[Service]\n", "[Service]\nRestart=always\n")))
t("oneshot + Restart=always -> SD04 high", ("high", "SD04") in codes(base.replace("[Service]\n", "[Service]\nType=oneshot\nRestart=always\n")))
t("After online without Wants -> SD05", ("medium", "SD05") in codes(base.replace("[Unit]\n", "[Unit]\nAfter=network-online.target\n")))
t("Wants online without After -> SD05", ("medium", "SD05") in codes(base.replace("[Unit]\n", "[Unit]\nWants=network-online.target\n")))
t("syslog.target obsolete -> SD05 low", ("low", "SD05") in codes(base.replace("[Unit]\n", "[Unit]\nAfter=syslog.target\n")))
t("typo directive -> SD08 with suggestion", any("did you mean RestartSec" in f[3] for f in R(base.replace("[Service]\n", "[Service]\nRestartSecs=5\n"))["findings"]))
t("invalid Restart value -> SD08 high", ("high", "SD08") in codes(base.replace("[Service]\n", "[Service]\nRestart=on-faliure\n")))
t("invalid value ignored, earlier valid one stays effective", R(base.replace("[Service]\n", "[Service]\nRestart=always\n"), "[Service]\nRestart=on-faliure\n")["restart"] == "always")
t("drop-in lines are valid key=value", all("=" in l or l.startswith(("#", "[")) for l in dropin(R(base)).splitlines()))
t("bad boolean -> SD08 medium", ("medium", "SD08") in codes(base.replace("[Service]\n", "[Service]\nPrivateTmp=maybe\n")))
t("StartLimitBurst in [Service] -> SD09", ("medium", "SD09") in codes(base.replace("[Service]\n", "[Service]\nStartLimitBurst=3\n")))
t("WantedBy in [Service] -> SD09", ("medium", "SD09") in codes(base.replace("[Service]\n", "[Service]\nWantedBy=multi-user.target\n")))
t("no [Install] -> SD11 medium", ("medium", "SD11") in codes("[Service]\nExecStart=/usr/bin/a\n"))
t("WantedBy=default.target -> SD11 low", ("low", "SD11") in codes(base.replace("multi-user.target", "default.target")))
t("duplicate scalar -> SD13", ("low", "SD13") in codes(base.replace("[Service]\n", "[Service]\nUser=a\nUser=b\n")))
t("drop-in override is not a duplicate", ("low", "SD13") not in codes(base.replace("[Service]\n", "[Service]\nUser=a\n"), "[Service]\nUser=b\n"))
t("drop-in value wins", R(base.replace("[Service]\n", "[Service]\nUser=a\n"), "[Service]\nUser=b\n")["user"] == "b")
t("ProtectHome with /home app -> SD14 high", ("high", "SD14") in codes(base.replace("/usr/bin/app", "/home/deploy/app/run").replace("[Service]\n", "[Service]\nProtectHome=yes\n")))
t("MDWE with node -> SD14 high", ("high", "SD14") in codes(base.replace("/usr/bin/app", "/usr/bin/node server.js").replace("[Service]\n", "[Service]\nMemoryDenyWriteExecute=yes\n")))
t("ProtectSystem=strict write path -> SD14", ("medium", "SD14") in codes(base.replace("/usr/bin/app", "/usr/bin/app --log-file=/var/log/app/app.log").replace("[Service]\n", "[Service]\nProtectSystem=strict\n")))
t("LogsDirectory clears strict write", ("medium", "SD14") not in codes(base.replace("/usr/bin/app", "/usr/bin/app --log-file=/var/log/app/app.log").replace("[Service]\n", "[Service]\nProtectSystem=strict\nLogsDirectory=app\n")))
t("CAP_SYS_ADMIN -> SD07 high", ("high", "SD07") in codes(base.replace("[Service]\n", "[Service]\nAmbientCapabilities=CAP_SYS_ADMIN\n")))
t("+ prefix -> SD01 medium", ("medium", "SD01") in codes(base.replace("[Service]\n", "[Service]\nUser=a\nExecStartPre=+/usr/bin/setup\n")))
t("TimeoutStopSec=infinity -> SD12", ("medium", "SD12") in codes(base.replace("[Service]\n", "[Service]\nTimeoutStopSec=infinity\n")))
t("continuation lines joined", R("[Service]\nExecStart=/usr/bin/app \\\n --port 80\nUser=a\n[Install]\nWantedBy=multi-user.target\n")["findings"] is not None and
parse("[Service]\nExecStart=/usr/bin/app \\\n --port 80\n", "x")["sections"]["Service"]["ExecStart"][0][0] == "/usr/bin/app --port 80")
t("bare exposure of minimal unit is 10", R(base)["exposure"] == 10.0)
t("ProtectSystem=full counts half", abs(R(base.replace("[Service]\n", "[Service]\nProtectSystem=full\n"))["exposure"] - 9.5) < 0.01)
try:
R("[Unit]\nDescription=x\n")
t("no [Service] rejected", False)
except InputError:
t("no [Service] rejected", True)
t("report has verdict and table", "**Verdict:" in report([R(base)]) and "| Sev | Code |" in report([R(base)]))
for n, ok in tests:
print("%s %s" % ("PASS" if ok else "FAIL", n))
p = sum(ok for _, ok in tests)
print("%d/%d passed" % (p, len(tests)))
return 0 if p == len(tests) else 1
def main(argv=None):
ap = argparse.ArgumentParser(description="Review systemd .service units for reliability and hardening.")
ap.add_argument("units", nargs="*", help="UNIT.service, optionally followed by drop-in .conf files; use ',' between separate units")
ap.add_argument("--json", action="store_true")
ap.add_argument("--dropin", action="store_true", help="print only the suggested hardening drop-in(s)")
ap.add_argument("--self-test", action="store_true")
a = ap.parse_args(argv)
if a.self_test:
return self_test()
if not a.units:
ap.error("give at least one .service file")
groups, cur = [], []
for u in a.units:
if u == ",":
groups.append(cur); cur = []
elif u.endswith(".service") and cur:
groups.append(cur); cur = [u]
else:
cur.append(u)
groups.append(cur)
try:
res = [run_files(g) for g in groups if g]
except InputError as e:
print("error: %s" % e, file=sys.stderr)
return 2
if a.json:
print(json.dumps([{**r, "findings": [dict(zip(("severity", "code", "where", "finding", "fix"), f)) for f in r["findings"]]} for r in res], indent=2))
elif a.dropin:
print("\n\n".join(dropin(r) for r in res))
else:
sys.stdout.write(report(res))
return 1 if any(f[0] == "high" for r in res for f in r["findings"]) else 0
if __name__ == "__main__":
sys.exit(main())