Install
openclaw skills install @thcjp/cron-guard-freeopenclaw skills install @thcjp/cron-guard-free让定时任务不脆弱。脚本优先、故障识别、防护栏、异常恢复,四层防护体系。
定时任务在生产环境中面临各种风险:路径错误、权限不足、依赖缺失、超时、资源竞争。本技能提供系统化的防护方案,让cron任务在故障发生时能够优雅处理而非崩溃。
┌─────────────────────────────────────────────────┐
│ 定时任务安全防护 (免费版) │
├─────────────────────────────────────────────────┤
│ │
│ ┌──────────────────────────────────────┐ │
│ │ 第一层:脚本优先原则 │ │
│ │ 命令行调用 > 内联代码 │ │
│ │ 独立脚本 > 嵌入逻辑 │ │
│ └──────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────┐ │
│ │ 第二层:故障模式识别 │ │
│ │ 路径错误 │ 权限不足 │ 依赖缺失 │ │
│ │ 超时 │ 资源竞争 │ 环境差异 │ │
│ └──────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────┐ │
│ │ 第三层:基础防护栏 │ │
│ │ 超时保护 │ 退出码检查 │ 日志记录 │ │
│ │ 错误通知 │ │
│ └──────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────┐ │
│ │ 第四层:异常恢复策略 │ │
│ │ 重试 │ 降级 │ 跳过 │ 告警 │ │
│ └──────────────────────────────────────┘ │
└─────────────────────────────────────────────────┘
import subprocess
import logging
from datetime import datetime
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
filename='cron_guard.log'
)
logger = logging.getLogger(__name__)
def run_script(script_path, timeout=300):
"""安全执行脚本,带超时和退出码检查"""
try:
logger.info(f"执行脚本:{script_path}")
result = subprocess.run(
["bash", script_path],
capture_output=True,
text=True,
timeout=timeout
)
if result.returncode == 0:
logger.info(f"脚本执行成功:{script_path}")
return True, result.stdout
else:
logger.error(f"脚本执行失败(退出码{result.returncode}):{script_path}")
logger.error(f"错误输出:{result.stderr}")
return False, result.stderr
except subprocess.TimeoutExpired:
logger.error(f"脚本超时({timeout}秒):{script_path}")
return False, "timeout"
except FileNotFoundError:
logger.error(f"脚本不存在:{script_path}")
return False, "not_found"
except Exception as e:
logger.error(f"执行异常:{e}")
return False, str(e)
success, output = run_script("/opt/scripts/backup.sh", timeout=600)
if not success:
print(f"执行失败,需要异常恢复")
配置完整防护栏与异常恢复:
详细代码示例已移至
references/detail.md
配置健康检查与故障模式库:
class FullCronGuard(CronGuard):
"""完整防护系统(含健康检查)"""
FAILURE_MODES = {
"path_error": {
"description": "脚本路径错误或文件不存在",
"symptom": "FileNotFoundError",
"prevention": "预检查文件存在性;使用绝对路径",
"recovery": "检查部署;修复路径"
},
"permission_denied": {
"description": "执行权限不足",
"symptom": "Permission denied",
"prevention": "预检查文件权限;chmod +x",
"recovery": "修复权限;检查运行用户"
},
"dependency_missing": {
"description": "依赖的命令或库缺失",
"symptom": "command not found / ImportError",
"prevention": "预检查依赖;锁定环境",
"recovery": "安装依赖;使用容器化"
},
"timeout": {
"description": "执行超时",
"symptom": "TimeoutExpired",
"prevention": "设置合理timeout;监控执行时长",
"recovery": "优化脚本;调大超时;降级执行"
},
"resource_contention": {
"description": "资源竞争(CPU/内存/磁盘)",
"symptom": "OOM / 磁盘满 / 响应慢",
"prevention": "资源监控;错峰执行",
"recovery": "释放资源;排队等待"
},
"env_diff": {
"description": "环境差异(PATH/变量不一致)",
"symptom": "crontab中能跑手动不行或反之",
"prevention": "脚本内设置环境变量;使用绝对路径",
"recovery": "统一环境;使用source"
}
}
def diagnose_failure(self, error_msg):
"""诊断故障模式"""
error_lower = error_msg.lower()
matches = []
for mode, info in self.FAILURE_MODES.items():
symptom = info["symptom"].lower()
if symptom in error_lower or any(
s in error_lower for s in symptom.split("/")
):
matches.append({
"mode": mode,
"description": info["description"],
"prevention": info["prevention"],
"recovery": info["recovery"]
})
if not matches:
matches.append({
"mode": "unknown",
"description": "未知故障",
"prevention": "加强日志记录",
"recovery": "人工排查"
})
return matches
def health_check(self, script_path):
"""任务健康检查"""
checks = {
"script_exists": Path(script_path).exists(),
"executable": Path(script_path).stat().st_mode & 0o111 if Path(script_path).exists() else False,
"recent_success": self._check_recent_success(script_path),
}
health_score = sum(checks.values()) / len(checks) * 100
status = "healthy" if health_score == 100 else "warning" if health_score >= 50 else "critical"
print(f"健康检查:{script_path}")
print(f" 文件存在:{'✓' if checks['script_exists'] else '✗'}")
print(f" 可执行: {'✓' if checks['executable'] else '✗'}")
print(f" 近期成功:{'✓' if checks['recent_success'] else '✗'}")
print(f" 健康度: {health_score:.0f}% ({status})")
return {"score": health_score, "status": status, "checks": checks}
def _check_recent_success(self, script_path):
"""检查近期是否有成功记录"""
import json
success_log = self.log_dir / "success.json"
if not success_log.exists():
return False
logs = json.loads(success_log.read_text(encoding="utf-8"))
recent = [l for l in logs if l.get("script") == script_path][-5:]
return any(l.get("status") == "success" for l in recent)
guard = FullCronGuard(log_dir="/tmp/cron_logs")
diagnosis = guard.diagnose_failure("Permission denied: /opt/scripts/backup.sh")
for d in diagnosis:
print(f"故障模式:{d['mode']}")
print(f" 描述:{d['description']}")
print(f" 预防:{d['prevention']}")
print(f" 恢复:{d['recovery']}")
guard.health_check("/opt/scripts/backup.sh")
| 原则 | 说明 | 示例 |
|---|---|---|
| 脚本优先 | 使用独立脚本而非内联命令 | bash /opt/scripts/backup.sh |
| 绝对路径 | 不依赖PATH变量 | /usr/bin/python3 而非 python3 |
| 环境自包含 | 脚本内设置所需环境 | export PATH=/usr/local/bin:$PATH |
| 幂等设计 | 重复执行不产生副作用 | 先检查再执行 |
输入: 用户提供脚本优先原则所需的指令和必要参数。 处理: 按照skill规范执行脚本优先原则操作,遵循单一意图原则。 输出: 返回脚本优先原则的执行结果,包含操作状态和输出数据。
| 模式 | 症状 | 预防 | 恢复 |
|---|---|---|---|
| 路径错误 | FileNotFoundError | 预检查;绝对路径 | 修复路径 |
| 权限不足 | Permission denied | chmod +x;检查用户 | 修复权限 |
| 依赖缺失 | command not found | 预检查依赖 | 安装依赖 |
| 超时 | TimeoutExpired | 合理timeout | 优化脚本 |
| 资源竞争 | OOM/磁盘满 | 资源监控 | 释放资源 |
| 环境差异 | crontab能跑手动不行 | 脚本内设置env | 统一环境 |
输入: 用户提供故障模式库所需的指令和必要参数。 处理: 按照skill规范执行故障模式库操作,遵循单一意图原则。 输出: 返回故障模式库的执行结果,包含操作状态和输出数据。
| 防护栏 | 说明 | 默认配置 |
|---|---|---|
| 超时保护 | 防止任务无限执行 | 300秒 |
| 退出码检查 | 非零退出码视为失败 | returncode==0 |
| 日志记录 | 记录执行过程和结果 | 文件日志 |
| 错误通知 | 失败时发送告警 | alerts.log |
| 预检查 | 执行前检查前置条件 | 路径+权限 |
| 重试机制 | 失败后自动重试 | 1次,间隔10秒 |
输入: 用户提供防护栏所需的指令和必要参数。 处理: 按照skill规范执行防护栏操作,遵循单一意图原则。 输出: 返回防护栏的执行结果,包含操作状态和输出数据。
| 策略 | 说明 | 适用场景 |
|---|---|---|
| 重试(retry) | 失败后重新执行 | 临时性故障 |
| 降级(degrade) | 使用备用方案 | 非关键路径 |
| 跳过(skip) | 跳过本次执行 | 非紧急任务 |
| 告警(alert) | 通知人工处理 | 关键任务 |
输入: 用户提供异常恢复策略所需的指令和必要参数。 处理: 按照skill规范执行异常恢复策略操作,遵循单一意图原则。 输出: 返回异常恢复策略的执行结果,包含操作状态和输出数据。 能力覆盖范围:本skill的核心能力覆盖以下场景关键词:cron、任务安全防护与异、常恢复免费版、含脚本优先原则、常见故障模式、基础防护栏、异常恢复、定时任务安全防护、免费版是面向、Agent、任务安全防护框架、让定时任务不脆弱、这一核心目标、提供防护栏设计、故障模式识别、异常恢复策略、确保定时任务在生、产环境中稳定可靠、Use、when、需要安全检测、合规审计、漏洞扫描、加密防护时使用、不适用于渗透测试、未授权目标、适用于独立开发者、企业团队和自动化、工作流场景等。
角色:运维工程师
场景描述:数据库备份脚本需要防护,防止超时或失败导致数据丢失。
guard = CronGuard(log_dir="/var/log/cron_guard")
guard.execute_guarded(
script_path="/opt/scripts/db_backup.sh",
timeout=3600, # 1小时超时
max_retries=2, # 最多重试2次
retry_delay=60, # 重试间隔1分钟
on_failure="alert" # 失败告警
)
角色:SRE工程师
场景描述:定时任务执行失败,需要快速定位故障原因。
guard = FullCronGuard(log_dir="/var/log/cron_guard")
diagnosis = guard.diagnose_failure("bash: python3: command not found")
角色:系统管理员
场景描述:定期检查定时任务的健康状态。
guard = FullCronGuard(log_dir="/var/log/cron_guard")
guard.health_check("/opt/scripts/cleanup.sh")
cron环境与交互式shell环境存在差异(PATH、环境变量等)。使用独立脚本而非内联命令有三个优势:(1) 脚本可以在开头设置所需环境,避免环境差异问题;(2) 脚本可以单独测试和调试,确保正确性;(3) 脚本可以被多个cron任务复用,减少重复。推荐所有cron任务都调用独立脚本。
这是典型的环境差异问题。cron环境的PATH最小化,很多命令找不到。解决方法:(1) 在脚本开头设置 export PATH=/usr/local/bin:/usr/bin:/bin;(2) 所有命令使用绝对路径(如 /usr/bin/python3);(3) 在脚本内 source ~/.bashrc 加载环境变量;(4) 检查cron运行用户与手动执行用户是否一致。
取决于任务类型:(1) 轻量任务(日志清理、状态检查):60-300秒;(2) 中量任务(数据同步、文件处理):300-1800秒;(3) 重量任务(数据库备份、大数据处理):1800-7200秒。原则是设置为预期执行时间的2-3倍,留出余量。
重试需要平衡可靠性和副作用:(1) 幂等任务(如查询、清理)可设置多次重试(3-5次);(2) 非幂等任务(如发送邮件、写入数据)建议最多1次重试;(3) 重试间隔建议递增(如10秒、30秒、60秒);(4) 连续失败超过阈值应触发告警而非无限重试。
使用健康检查功能检查三个维度:(1) 脚本文件是否存在且可执行;(2) 近期是否有成功执行记录;(3) 退出码是否正常。健康度100%为healthy,50%以上为warning,低于50%为critical。建议定期运行健康检查,及时发现潜在问题。
| 依赖项 | 类型 | 是否必需 | 获取方式 |
|---|---|---|---|
| Python标准库 | 内置 | 必需 | Python自带(subprocess/logging/json/pathlib) |
| bash | 系统命令 | 必需 | Linux/macOS自带;Windows需WSL或Git Bash |
| LLM API | API | 必需 | 由Agent平台内置LLM提供 |
本技能基于原始开源定时任务防护作品改进,保留原始版权声明:
本改进作品在原始作品基础上进行了深度差异化改造,包括但不限于:
原始MIT license允许使用、复制、修改和分发,需保留版权声明。本改进作品在保留原始版权声明的基础上添加自有署名,完全符合MIT license要求。
本免费体验版限制以下高级功能:
解锁全部功能请使用专业版:cron-guard-pro
| 错误场景 | 原因 | 处理方式 |
|---|---|---|
| 配置错误 | 参数缺失或格式错误 | 检查依赖说明中的配置要求 |
| 运行时错误 | 运行环境不满足 | 确认运行环境符合依赖说明 |
| 网络错误 | 连接超时或不可达 | 执行ping命令测试网络连通性,检查防火墙和代理设置连接后执行ping命令测试网络连通性,检查防火墙和代理设置连接后重新执行命令,参考国内替代方案 |
输入:用户提供操作指令和必要参数
输出:返回执行结果,包含操作状态和输出数据
用户: 执行核心功能
Skill: 正在执行核心功能...
Skill: 执行完成,结果如下: 操作成功