296 lines
17 KiB
Python
296 lines
17 KiB
Python
"""CLI entry point."""
|
||
|
||
import argparse
|
||
import sys
|
||
|
||
from . import __version__, commands
|
||
from .client import GraylogClient
|
||
from .config import ConfigError, load_config
|
||
from .http import GraylogError, HttpClient
|
||
|
||
EPILOG = """\
|
||
示例:
|
||
# 1) 先确认连通性、权限和这个版本支持哪些接口
|
||
graylog-ops --url http://edu-master:9000 --token $GRAYLOG_TOKEN probe
|
||
|
||
# 2) 摸清"客户端 IP 到底在哪个字段"（最关键的一步）
|
||
graylog-ops discover --range 3600 --stream 业务访问日志
|
||
|
||
# 3) 按条件看 top IP，判断阈值是否合理
|
||
graylog-ops top-ips --field client_ip --query 'status:401' --range 300 --size 30
|
||
|
||
# 3.5) 生成 profile，然后做封禁判断（只读，dry-run 语义）
|
||
graylog-ops discover --range 3600 --save-profile profiles/edu.json
|
||
graylog-ops judge --profile profiles/edu.json --range 300 --allow allowlist.txt --explain 5
|
||
|
||
# 4) 把环境信息整体落盘，便于离线分析/复盘
|
||
graylog-ops snapshot --dir snapshots/$(date +%F) --sample 500
|
||
|
||
# 5) 探测未知接口（版本差异时用）
|
||
graylog-ops raw /api/system/inputs
|
||
graylog-ops raw /api/search/universal/terms --param field=source --param range=60
|
||
|
||
# 6) 一次性产出整套信息（probe/discover/judge/snapshot），便于离线分析
|
||
graylog-ops collect --dir collect-2026-10-18 --range 3600 --allow allowlist.example.txt
|
||
|
||
环境变量: GRAYLOG_URL GRAYLOG_TOKEN GRAYLOG_USER GRAYLOG_PASSWORD
|
||
GRAYLOG_INSECURE GRAYLOG_TIMEOUT GRAYLOG_RETRIES
|
||
配置文件: ~/.config/graylog-ops/config.json 或 ./graylog-ops.json
|
||
"""
|
||
|
||
|
||
def _add_common(parser, suppress):
|
||
"""Add connection/output flags.
|
||
|
||
With ``suppress=True`` the defaults are SUPPRESS so a sub-parser never
|
||
clobbers a value that was already parsed by the root parser.
|
||
"""
|
||
|
||
def default(value):
|
||
return argparse.SUPPRESS if suppress else value
|
||
|
||
parser.add_argument("--url", metavar="URL", default=default(None),
|
||
help="Graylog 地址，如 http://edu-master:9000")
|
||
parser.add_argument("--token", metavar="TOKEN", default=default(None),
|
||
help="API token（推荐；设置后忽略 --user/--password）")
|
||
parser.add_argument("--user", metavar="USER", default=default(None),
|
||
help="用户名（配合 --password）")
|
||
parser.add_argument("--password", metavar="PASSWORD", default=default(None),
|
||
help="密码")
|
||
parser.add_argument("--insecure", action="store_true",
|
||
default=argparse.SUPPRESS if suppress else False,
|
||
help="跳过 HTTPS 证书校验")
|
||
parser.add_argument("--timeout", type=float, metavar="SECONDS", default=default(None),
|
||
help="单请求超时，默认 30")
|
||
parser.add_argument("--retries", type=int, metavar="N", default=default(None),
|
||
help="失败重试次数，默认 2")
|
||
parser.add_argument("--config", metavar="PATH", default=default(None),
|
||
help="JSON 配置文件路径")
|
||
parser.add_argument("--json", action="store_true",
|
||
default=argparse.SUPPRESS if suppress else False,
|
||
help="输出 JSON（便于管道/脚本消费）")
|
||
parser.add_argument("--out", metavar="PATH", default=default(None),
|
||
help="把结果 JSON 写入文件")
|
||
parser.add_argument("-v", "--verbose", action="store_true",
|
||
default=argparse.SUPPRESS if suppress else False,
|
||
help="打印每个 HTTP 请求")
|
||
|
||
|
||
def build_parser():
|
||
common = argparse.ArgumentParser(add_help=False)
|
||
_add_common(common, suppress=True)
|
||
|
||
parser = argparse.ArgumentParser(
|
||
prog="graylog-ops",
|
||
description="Graylog 4.0 只读信息采集工具（为 IP 封禁规则摸底）",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog=EPILOG,
|
||
)
|
||
_add_common(parser, suppress=False)
|
||
parser.add_argument("--version", action="version", version="graylog-ops %s" % __version__)
|
||
|
||
sub = parser.add_subparsers(dest="command", metavar="<command>")
|
||
|
||
def add(name, handler, help_text):
|
||
cmd = sub.add_parser(name, parents=[common], help=help_text, description=help_text)
|
||
cmd.set_defaults(func=handler)
|
||
return cmd
|
||
|
||
add("probe", commands.cmd_probe, "检查连通性、认证、接口可用性矩阵")
|
||
add("system", commands.cmd_system, "节点/索引器/版本/license 概要")
|
||
add("streams", commands.cmd_streams, "列出 stream（分流与权限边界）").add_argument(
|
||
"--rules", action="store_true", help="同时拉取每个 stream 的规则（N+1 请求）")
|
||
add("inputs", commands.cmd_inputs, "列出 input 及监听地址")
|
||
add("fields", commands.cmd_fields, "列出 Graylog 已注册字段与类型")
|
||
add("indices", commands.cmd_indices, "index set 与索引大小/保留策略")
|
||
add("alerts", commands.cmd_alerts, "现有事件定义与通知方式（改造前先看）").add_argument(
|
||
"--limit", type=int, default=None, help="最多返回条数")
|
||
|
||
discover = add("discover", commands.cmd_discover, "采样日志，推断真实客户端 IP 字段")
|
||
discover.add_argument("--stream", help="stream id 或标题子串")
|
||
discover.add_argument("--query", default="*", help="查询条件，默认 *")
|
||
discover.add_argument("--range", type=int, default=3600, help="时间窗（秒），默认 3600")
|
||
discover.add_argument("--sample", type=int, default=500, help="采样消息数，默认 500")
|
||
discover.add_argument("--save-profile", metavar="PATH",
|
||
help="把推断出的字段映射写成 profile，供 judge 使用")
|
||
|
||
judge = add("judge", commands.cmd_judge,
|
||
"把聚合信号判成封禁候选（只读，不执行封禁）")
|
||
judge.add_argument("--profile", metavar="PATH", help="discover --save-profile 生成的文件")
|
||
judge.add_argument("--range", type=int, default=300, help="时间窗（秒），默认 300")
|
||
judge.add_argument("--query", default="*", help="限定查询范围，默认 *")
|
||
judge.add_argument("--stream", help="stream id 或标题子串")
|
||
judge.add_argument("--field", help="客户端 IP 字段（覆盖 profile）")
|
||
judge.add_argument("--status-field", help="状态码字段")
|
||
judge.add_argument("--uri-field", help="URI 字段")
|
||
judge.add_argument("--ua-field", help="User-Agent 字段")
|
||
judge.add_argument("--method-field", help="HTTP 方法字段")
|
||
judge.add_argument("--allow", action="append", metavar="FILE",
|
||
help="白名单文件（IP/CIDR，可重复）")
|
||
judge.add_argument("--include-internal", action="store_true",
|
||
help="允许内网地址进入候选（默认排除）")
|
||
judge.add_argument("--min-requests", type=int, help="低于该请求数的 IP 不参与判分")
|
||
judge.add_argument("--min-score", type=float, help="低于该分数的候选不输出")
|
||
judge.add_argument("--top", type=int, default=30, help="最多输出条数，默认 30")
|
||
judge.add_argument("--size", type=int, default=200, help="每类聚合返回的 IP 数上限，默认 200")
|
||
judge.add_argument("--verify", type=int, default=0, metavar="N",
|
||
help="对前 N 个候选反查 URI 集中度（识别自动化客户端，每候选多 1 次查询）")
|
||
judge.add_argument("--explain", type=int, default=0, metavar="N",
|
||
help="为前 N 个候选附带样例消息")
|
||
judge.add_argument("--all", action="store_true", help="输出全部候选（含 ignore）")
|
||
|
||
watch = add("watch", commands.cmd_watch,
|
||
"持续检测恶意公网 IP + 本地 Web 面板（SQLite 持久化）")
|
||
watch.add_argument("--profile", metavar="PATH", help="字段映射 profile")
|
||
watch.add_argument("--stream", help="只检测某个 stream")
|
||
watch.add_argument("--interval", type=int, default=60, help="检测间隔（秒），默认 60")
|
||
watch.add_argument("--window", type=int, default=900, help="检测时间窗（秒），默认 900")
|
||
watch.add_argument("--host", default="127.0.0.1", help="面板监听地址")
|
||
watch.add_argument("--port", type=int, default=8088, help="面板端口，默认 8088")
|
||
watch.add_argument("--db", default="data/ips.db", help="SQLite 路径")
|
||
watch.add_argument("--backfill-threshold", type=int, default=1800,
|
||
help="启动时检测缺口超过多少秒自动补测（默认 1800）")
|
||
watch.add_argument("--backfill-max-gap", type=int, default=7 * 86400,
|
||
help="自动补测允许的最大缺口秒数（默认 7 天）")
|
||
watch.add_argument("--no-auto-backfill", action="store_true",
|
||
help="启动时不自动补测历史缺口")
|
||
watch.add_argument("--behavior-interval", type=int, default=900,
|
||
help="行为检测刷新间隔（秒，默认 900）")
|
||
watch.add_argument("--no-behavior", action="store_true",
|
||
help="关闭行为检测线程，只跑特征检测")
|
||
watch.add_argument("--notify-config", default="notify.json",
|
||
help="新增恶意 IP 的 webhook 配置（不存在则不推送）")
|
||
watch.add_argument("--notify-test", action="store_true",
|
||
help="发送一条测试推送后退出")
|
||
watch.add_argument("--asn-db", default="data/geo/asn.db",
|
||
help="ASN 库路径（空字符串禁用）")
|
||
watch.add_argument("--geo-db", default="data/geo/geo.db",
|
||
help="DB-IP Lite 生成的离线地理库路径（空字符串禁用）")
|
||
watch.add_argument("--path-prefix", metavar="PATH", default="",
|
||
help="只在该路径前缀下提供服务，根路径返回 404（用于对外分享）")
|
||
watch.add_argument("--forget", type=int, default=7200, help="多久没再出现就从列表移除（秒）")
|
||
watch.add_argument("--allow", action="append", metavar="FILE", help="白名单文件，可重复")
|
||
watch.add_argument("--field", help="客户端 IP 字段")
|
||
watch.add_argument("--status-field", help="状态码字段")
|
||
watch.add_argument("--uri-field", help="URI 字段")
|
||
watch.add_argument("--ua-field", help="User-Agent 字段")
|
||
watch.add_argument("--min-requests", type=int, default=3, help="低于该请求数不参与判定")
|
||
watch.add_argument("--size", type=int, default=200, help="每类聚合返回的 IP 数上限")
|
||
watch.add_argument("--verify", type=int, default=8, help="对前 N 个候选做 URI 核验")
|
||
watch.add_argument("--min-observe-score", type=float, default=10.0,
|
||
help="观察级候选低于该分数不入库，默认 10")
|
||
watch.add_argument("--top", type=int, default=50, help="--once 模式下打印的条数")
|
||
watch.add_argument("--export", metavar="PATH",
|
||
help="把高置信恶意 IP 导出为纯文本名单（排除疑似 NAT）")
|
||
watch.add_argument("--once", action="store_true", help="只跑一轮后退出（用于离线分析）")
|
||
watch.add_argument("--warm-limit", type=int, default=5,
|
||
help="每轮为多少个恶意 IP 预生成证据快照，默认 5")
|
||
watch.add_argument("--fix-times", "--fix-first-seen", dest="fix_first_seen",
|
||
action="store_true",
|
||
help="回查日志，修正首次攻击 / 最近攻击 / 最近出现时间")
|
||
watch.add_argument("--warm-only", action="store_true",
|
||
help="只预热证据快照后退出（配合 --warm-limit）")
|
||
watch.add_argument("--serve-only", action="store_true",
|
||
help="只提供面板，不做检测（用于浏览已有数据库）")
|
||
|
||
collect = add("collect", commands.cmd_collect,
|
||
"一条命令产出 probe/discover/judge/snapshot 整套信息")
|
||
collect.add_argument("--dir", help="输出目录，默认 graylog-collect-<时间戳>")
|
||
collect.add_argument("--range", type=int, default=3600, help="discover 的时间窗（秒）")
|
||
collect.add_argument("--judge-range", type=int, default=300, help="judge 的时间窗（秒）")
|
||
collect.add_argument("--stream", help="只针对某个 stream（传给 discover/judge）")
|
||
collect.add_argument("--allow", action="append", metavar="FILE", help="白名单文件，可重复")
|
||
|
||
for name, default_field in (("terms", None), ("top-ips", "client_ip")):
|
||
cmd = add(name, commands.cmd_terms,
|
||
"按字段做 terms 聚合（top-ips 是 field=client_ip 的快捷方式）")
|
||
cmd.add_argument("--field", required=(default_field is None), default=default_field,
|
||
help="聚合字段")
|
||
cmd.add_argument("--query", default="*", help="查询条件，默认 *")
|
||
cmd.add_argument("--range", type=int, default=300, help="时间窗（秒），默认 300")
|
||
cmd.add_argument("--size", type=int, default=50, help="返回条数，默认 50")
|
||
cmd.add_argument("--min-count", type=int, default=None, help="过滤低于该计数的项")
|
||
cmd.add_argument("--stream", help="stream id 或标题子串")
|
||
cmd.add_argument("--from", dest="from_ts", metavar="ISO8601", help="绝对时间起（UTC）")
|
||
cmd.add_argument("--to", dest="to_ts", metavar="ISO8601", help="绝对时间止（UTC）")
|
||
|
||
messages = add("messages", commands.cmd_messages, "拉取原始消息（JSON/表格/JSONL）")
|
||
messages.add_argument("--query", default="*", help="查询条件，默认 *")
|
||
messages.add_argument("--range", type=int, default=300, help="时间窗（秒），默认 300")
|
||
messages.add_argument("--limit", type=int, default=50, help="条数，默认 50")
|
||
messages.add_argument("--offset", type=int, default=0, help="偏移")
|
||
messages.add_argument("--fields", help="只取这些字段，逗号分隔")
|
||
messages.add_argument("--stream", help="stream id 或标题子串")
|
||
messages.add_argument("--jsonl", action="store_true", help="每行一个 JSON 消息")
|
||
|
||
histogram = add("histogram", commands.cmd_histogram, "消息量时间分布（估算阈值/容量）")
|
||
histogram.add_argument("--query", default="*", help="查询条件，默认 *")
|
||
histogram.add_argument("--range", type=int, default=3600, help="时间窗（秒），默认 3600")
|
||
histogram.add_argument("--interval", default="1m", help="时间桶，如 1m/5m/1h")
|
||
histogram.add_argument("--stream", help="stream id 或标题子串")
|
||
|
||
snapshot = add("snapshot", commands.cmd_snapshot, "整体环境快照落到目录（每个资源一个 json）")
|
||
snapshot.add_argument("--dir", help="输出目录，默认 graylog-snapshot-<时间戳>")
|
||
snapshot.add_argument("--sample", type=int, default=0, help="附带采样消息条数，默认 0")
|
||
|
||
raw = add("raw", commands.cmd_raw, "直接调用任意 API 路径（探索未封装的接口）")
|
||
raw.add_argument("path", help="如 /api/system/inputs")
|
||
raw.add_argument("--method", default="GET", help="HTTP 方法，默认 GET")
|
||
raw.add_argument("--body", metavar="JSON", help="JSON 请求体，或 @文件路径")
|
||
raw.add_argument("--param", action="append", metavar="K=V", help="查询参数，可重复")
|
||
raw.add_argument("--text", action="store_true", help="按文本返回（非 JSON）")
|
||
|
||
return parser
|
||
|
||
|
||
def main(argv=None):
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
if not getattr(args, "command", None):
|
||
parser.print_help()
|
||
return 2
|
||
|
||
try:
|
||
config = load_config(args)
|
||
except ConfigError as exc:
|
||
print("配置错误: %s" % exc, file=sys.stderr)
|
||
return 2
|
||
|
||
http_client = HttpClient(
|
||
config.url,
|
||
username=config.credentials[0],
|
||
password=config.credentials[1],
|
||
timeout=config.timeout,
|
||
retries=config.retries,
|
||
insecure=config.insecure,
|
||
verbose=bool(getattr(args, "verbose", False)),
|
||
)
|
||
ctx = commands.Context(args, config, GraylogClient(http_client))
|
||
|
||
try:
|
||
return args.func(ctx, args) or 0
|
||
except GraylogError as exc:
|
||
print("Graylog API 错误: %s" % exc, file=sys.stderr)
|
||
if exc.status == 401:
|
||
print(
|
||
"提示: token 认证是把 token 当用户名、字面量 token 当密码；"
|
||
"确认 token 未吊销且账号有对应权限",
|
||
file=sys.stderr,
|
||
)
|
||
elif exc.status == 403:
|
||
print("提示: 账号缺少该资源权限（如 messages:read / searches:*）", file=sys.stderr)
|
||
elif exc.status == 404:
|
||
print(
|
||
"提示: 该版本可能没有这个接口。先跑 `probe` 看可用性，"
|
||
"或用 `raw /api/...` 直接验证",
|
||
file=sys.stderr,
|
||
)
|
||
elif exc.status in (400, 422):
|
||
print("提示: 参数不被该版本接受，加 -v 查看实际请求的 URL", file=sys.stderr)
|
||
return 1
|
||
except BrokenPipeError:
|
||
return 0
|
||
except KeyboardInterrupt:
|
||
print("", file=sys.stderr)
|
||
return 130
|