Opinionated conventions for writing safe, readable and maintainable Bash scripts.
# 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.70Forecasts when disks and volumes will fill up from usage samples (CSV of timestamp, volume, used and size). Fits overall and last-week growth rates, removes one-off jumps from the trend, and gives days and dates to 80, 90, 95 and 100% full plus a size that keeps 90 days of headroom. Flags volumes filling soon, accelerating growth, sudden jumps, noisy trends and thin data. Includes a stdlib Python forecaster with --self-test.
---
name: disk-capacity-runway-forecaster
description: Forecasts when disks and volumes will fill up from usage samples (CSV of timestamp, volume, used and size). Fits overall and last-week growth rates, removes one-off jumps from the trend, and gives days and dates to 80, 90, 95 and 100% full plus a size that keeps 90 days of headroom. Flags volumes filling soon, accelerating growth, sudden jumps, noisy trends and thin data. Includes a stdlib Python forecaster with --self-test.
---
# 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())Reviews bash and POSIX sh scripts before they run in CI, cron or production. Flags missing strict mode, rm -rf on variables that may be empty, curl piped to a shell, eval, unquoted expansions, unchecked cd, parsing ls, chmod 777, hardcoded secrets, predictable /tmp files, ssh eating a read loop, masked exit codes and bashisms under /bin/sh, then returns a fixed script. Includes a stdlib Python linter with --self-test.
--- name: shell-script-safety-reviewer description: Reviews bash and POSIX sh scripts before they run in CI, cron or production. Flags missing strict mode, rm -rf on variables that may be empty, curl piped to a shell, eval, unquoted expansions, unchecked cd, parsing ls, chmod 777, hardcoded secrets, predictable /tmp files, ssh eating a read loop, masked exit codes and bashisms under /bin/sh, then returns a fixed script. Includes a stdlib Python linter with --self-test. --- # 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())