Hermes Agent
4a1d949e41
完整 Web 管理平台,涵盖: P0 生产加固: - 会话超时(空闲+绝对) + CSRF 防护 + 登录失败限流 - HTTPS/TLS 证书管理(自签生成/上传/删除/过期提醒) - SSL Bump (HTTPS 透明代理) 可视化配置 - 配置文件 diff 预览 + 二次确认才写入 - 服务控制二次确认 + 操作期间按钮锁定 P1 运维能力: - 日志轮转管理(logrotate) + 日志状态监控 - 告警规则(5xx/命中率/磁盘/客户端流量) - 多 Squid 实例管理 + 一键切换 - squidclient mgr 实时性能指标 - 流量异常检测(突增/大文件/高频小包/新客户端) P2 日志分析增强: - 日志持久化到 SQLite + 历史时间范围查询 - 导出 CSV/JSON(日志/KPI/异常/审计) - GeoIP 地图(Leaflet + ip-api.com 离线可选 GeoLite2) - 每客户端 24h 趋势图 - 自定义 Squid logformat 解析器 P3 体验锦上添花: - 暗/亮主题切换(cookie 持久化) - WebSSH 终端(xterm.js + paramiko) - 完整审计日志 + 用户管理 技术栈: Flask 3.0 + SQLAlchemy 2.0 + SQLite + Chart.js + Leaflet 总代码量: ~16500 行 (15 Python 模块 + 34 模板 + CSS) 路由数: 73
496 lines
16 KiB
Python
496 lines
16 KiB
Python
"""TLS / HTTPS certificate management for Squid Web Manager.
|
|
|
|
Self-contained helper around the system ``openssl`` binary. We deliberately
|
|
avoid Python ``cryptography`` / ``pyOpenSSL`` to keep the dependency surface
|
|
small; everything goes through ``subprocess``.
|
|
|
|
Public API (all return a tuple ``(ok: bool, message: str, data: dict)``
|
|
on failure / success paths so callers can ``flash()`` directly):
|
|
|
|
- ``is_openssl_available()`` -> ``(ok, msg, data)`` with ``data["openssl"]`` path
|
|
- ``generate_self_signed(common_name, days, cert_dir)``
|
|
- ``parse_pem_info(cert_path)``
|
|
- ``list_certificates(cert_dir)``
|
|
- ``upload_certificate(uploaded_file, cert_dir)``
|
|
- ``delete_certificate(name, cert_dir)``
|
|
|
|
Security notes:
|
|
- File names are sanitised; ``..`` and path separators are rejected.
|
|
- Uploads are limited to 1 MiB and only ``.pem`` / ``.crt`` / ``.key`` accepted.
|
|
- Self-signed filenames embed CN + date + short hash, no spaces.
|
|
- Audit callers MUST NOT pass private-key material; only metadata + hash.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
# --- Limits / allowed extensions --------------------------------------------
|
|
|
|
MAX_UPLOAD_BYTES = 1 * 1024 * 1024 # 1 MiB
|
|
ALLOWED_EXTENSIONS = (".pem", ".crt", ".key")
|
|
SAFE_NAME_RE = re.compile(r"^[A-Za-z0-9_.\-]+$")
|
|
# Allow CN to contain dots/underscores but normalise whitespace to underscores.
|
|
CN_SAFE_RE = re.compile(r"[^A-Za-z0-9._-]")
|
|
|
|
|
|
# --- helpers ----------------------------------------------------------------
|
|
|
|
|
|
def _err(message: str, **extra) -> tuple[bool, str, dict]:
|
|
data = {"error": message}
|
|
data.update(extra)
|
|
return False, message, data
|
|
|
|
|
|
def _ok(message: str = "OK", **extra) -> tuple[bool, str, dict]:
|
|
data = {"ok": True}
|
|
data.update(extra)
|
|
return True, message, data
|
|
|
|
|
|
def _short_hash(data: bytes) -> str:
|
|
"""Return first 8 hex chars of SHA-256 for filename uniqueness."""
|
|
return hashlib.sha256(data).hexdigest()[:8]
|
|
|
|
|
|
def _safe_filename(name: str) -> str:
|
|
"""Strip directory components and reject anything non-ASCII-clean."""
|
|
if not name:
|
|
return ""
|
|
base = os.path.basename(name)
|
|
if base != name:
|
|
return ""
|
|
if not SAFE_NAME_RE.match(base):
|
|
return ""
|
|
return base
|
|
|
|
|
|
def _ensure_dir(cert_dir: str) -> tuple[bool, str, dict]:
|
|
"""Make sure ``cert_dir`` exists and is writable."""
|
|
if not cert_dir:
|
|
return _err("cert_dir 不能为空")
|
|
try:
|
|
os.makedirs(cert_dir, exist_ok=True)
|
|
except OSError as e:
|
|
return _err(f"无法创建目录 {cert_dir}: {e}")
|
|
if not os.path.isdir(cert_dir):
|
|
return _err(f"{cert_dir} 不是目录")
|
|
return _ok("dir-ok", dir=cert_dir)
|
|
|
|
|
|
def _safe_join(cert_dir: str, name: str) -> tuple[str | None, str | None]:
|
|
"""Join ``cert_dir`` + ``name`` and reject path traversal."""
|
|
clean = _safe_filename(name)
|
|
if not clean:
|
|
return None, "文件名包含非法字符 (只允许字母/数字/_-.)"
|
|
full = os.path.abspath(os.path.join(cert_dir, clean))
|
|
base = os.path.abspath(cert_dir)
|
|
# Use os.path.commonpath for traversal rejection.
|
|
try:
|
|
if os.path.commonpath([full, base]) != base:
|
|
return None, "不允许的路径"
|
|
except ValueError:
|
|
return None, "不允许的路径"
|
|
return full, None
|
|
|
|
|
|
# --- public API --------------------------------------------------------------
|
|
|
|
|
|
def is_openssl_available() -> tuple[bool, str, dict]:
|
|
"""Return (ok, message, data) where data["openssl"] is the binary path."""
|
|
path = shutil.which("openssl")
|
|
if not path:
|
|
return _err("系统未安装 openssl 或不在 PATH 中")
|
|
try:
|
|
r = subprocess.run(
|
|
[path, "version"], capture_output=True, text=True, timeout=5
|
|
)
|
|
if r.returncode != 0:
|
|
return _err(f"openssl 调用失败: {(r.stderr or '').strip()}")
|
|
version = (r.stdout or "").strip()
|
|
except (subprocess.TimeoutExpired, OSError) as e:
|
|
return _err(f"无法执行 openssl: {e}")
|
|
return _ok("openssl 可用", openssl=path, version=version)
|
|
|
|
|
|
def generate_self_signed(
|
|
common_name: str, days: int = 365, cert_dir: str = ""
|
|
) -> tuple[bool, str, dict]:
|
|
"""Generate a self-signed PEM cert+key with combined contents.
|
|
|
|
Output filename: ``{CN}_{YYYYMMDD}_{short_hash8}.pem`` containing both
|
|
certificate and private key sections, parsable by OpenSSL directly.
|
|
"""
|
|
available_ok, _, avail_data = is_openssl_available()
|
|
if not available_ok:
|
|
return _err(avail_data.get("error", "openssl 不可用"))
|
|
openssl = avail_data["openssl"]
|
|
|
|
ensured_ok, _, ensured_data = _ensure_dir(cert_dir)
|
|
if not ensured_ok:
|
|
return False, ensured_data["error"], ensured_data
|
|
|
|
cn = (common_name or "").strip()
|
|
if not cn:
|
|
return _err("Common Name (CN) 不能为空")
|
|
# File-system safe CN for filename use only; the certificate itself
|
|
# preserves the original CN via ``-subj``.
|
|
safe_cn = CN_SAFE_RE.sub("_", cn).strip("._-")
|
|
if not safe_cn:
|
|
return _err("CN 经清洗后为空,仅含非法字符")
|
|
if len(safe_cn) > 64:
|
|
safe_cn = safe_cn[:64].rstrip("._-")
|
|
|
|
try:
|
|
days_i = int(days)
|
|
except (TypeError, ValueError):
|
|
return _err("days 必须是整数")
|
|
if days_i < 1 or days_i > 3650:
|
|
return _err("days 必须在 1 ~ 3650 之间")
|
|
|
|
date_part = datetime.now().strftime("%Y%m%d")
|
|
# 8-char random hash ensures repeated runs with same CN don't collide.
|
|
rand_hash = hashlib.sha256(
|
|
f"{cn}|{datetime.now().timestamp()}|{os.urandom(8).hex()}".encode()
|
|
).hexdigest()[:8]
|
|
out_name = f"{safe_cn}_{date_part}_{rand_hash}.pem"
|
|
out_path = os.path.join(cert_dir, out_name)
|
|
|
|
# Escape the CN for the -subj argument: /CN=something
|
|
# Allow only printable ASCII in subject to avoid openssl quoting issues.
|
|
subj_cn = re.sub(r"[^\x20-\x7e]", "_", cn)
|
|
if "/" in subj_cn:
|
|
# '/' starts a new RDN component in -subj; replace with '_'
|
|
subj_cn = subj_cn.replace("/", "_")
|
|
subject = f"/CN={subj_cn}"
|
|
|
|
cmd = [
|
|
openssl, "req", "-x509", "-newkey", "rsa:2048",
|
|
"-keyout", out_path,
|
|
"-out", out_path,
|
|
"-days", str(days_i),
|
|
"-nodes",
|
|
"-subj", subject,
|
|
]
|
|
try:
|
|
r = subprocess.run(
|
|
cmd, capture_output=True, text=True, timeout=30
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
# best-effort cleanup of half-written file
|
|
if os.path.exists(out_path):
|
|
try:
|
|
os.remove(out_path)
|
|
except OSError:
|
|
pass
|
|
return _err("openssl 生成超时 (>30s)")
|
|
except OSError as e:
|
|
return _err(f"执行 openssl 失败: {e}")
|
|
|
|
if r.returncode != 0:
|
|
# clean any half-written file
|
|
if os.path.exists(out_path):
|
|
try:
|
|
os.remove(out_path)
|
|
except OSError:
|
|
pass
|
|
err = (r.stderr or r.stdout or "").strip()
|
|
return _err(f"openssl 失败: {err[:400]}")
|
|
|
|
if not os.path.isfile(out_path):
|
|
return _err(f"openssl 未生成文件: {out_path}")
|
|
|
|
# Validate the cert parses back.
|
|
parse_ok, parse_msg, parse_data = parse_pem_info(out_path)
|
|
if not parse_ok:
|
|
return False, f"生成的证书无法解析: {parse_msg}", parse_data
|
|
|
|
# Audit-friendly fingerprint of the cert portion only (no private key).
|
|
fingerprint = parse_data.get("sha256", "")
|
|
return _ok(
|
|
f"已生成自签名证书: {out_name}",
|
|
name=out_name,
|
|
path=out_path,
|
|
cn=parse_data.get("cn", cn),
|
|
days=days_i,
|
|
expires=parse_data.get("not_after", ""),
|
|
fingerprint=fingerprint,
|
|
size=os.path.getsize(out_path),
|
|
)
|
|
|
|
|
|
def parse_pem_info(cert_path: str) -> tuple[bool, str, dict]:
|
|
"""Parse a PEM file and return subject CN, validity, sha256 fingerprint.
|
|
|
|
If the file is not a recognised PEM certificate (e.g. a private key)
|
|
``ok`` is False and the data dict explains why.
|
|
"""
|
|
if not cert_path or not os.path.isfile(cert_path):
|
|
return _err(f"找不到证书: {cert_path}")
|
|
|
|
# Subject (CN)
|
|
cn = ""
|
|
try:
|
|
r = subprocess.run(
|
|
["openssl", "x509", "-noout", "-subject", "-nameopt", "RFC2253",
|
|
"-in", cert_path],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
if r.returncode != 0:
|
|
return _err(
|
|
"openssl 无法解析 (文件可能不是证书, 或包含私钥): "
|
|
+ (r.stderr or "").strip()[:200]
|
|
)
|
|
# subject= /CN=foo
|
|
subj_line = (r.stdout or "").strip()
|
|
m = re.search(r"CN\s*=\s*([^,/\n]+)", subj_line)
|
|
if m:
|
|
cn = m.group(1).strip()
|
|
except (subprocess.TimeoutExpired, OSError) as e:
|
|
return _err(f"openssl x509 调用失败: {e}")
|
|
|
|
# Validity: notBefore / notAfter
|
|
not_before = ""
|
|
not_after = ""
|
|
try:
|
|
r = subprocess.run(
|
|
["openssl", "x509", "-noout", "-dates", "-in", cert_path],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
if r.returncode == 0:
|
|
for line in (r.stdout or "").splitlines():
|
|
if line.startswith("notBefore="):
|
|
not_before = line.split("=", 1)[1].strip()
|
|
elif line.startswith("notAfter="):
|
|
not_after = line.split("=", 1)[1].strip()
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
pass
|
|
|
|
# Serial
|
|
serial = ""
|
|
try:
|
|
r = subprocess.run(
|
|
["openssl", "x509", "-noout", "-serial", "-in", cert_path],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
if r.returncode == 0:
|
|
m = re.search(r"=([0-9A-Fa-f]+)", r.stdout or "")
|
|
if m:
|
|
serial = m.group(1)
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
pass
|
|
|
|
# SHA-256 fingerprint
|
|
sha256 = ""
|
|
try:
|
|
r = subprocess.run(
|
|
["openssl", "x509", "-noout", "-fingerprint", "-sha256",
|
|
"-in", cert_path],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
if r.returncode == 0:
|
|
# output e.g. "sha256 Fingerprint=AA:BB:..."
|
|
line = (r.stdout or "").strip()
|
|
if "=" in line:
|
|
sha256 = line.split("=", 1)[1].replace(":", "").lower()
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
pass
|
|
|
|
days_remaining = None
|
|
expires_iso = ""
|
|
status = "unknown"
|
|
if not_after:
|
|
try:
|
|
# notAfter format: e.g. "Jun 14 12:00:00 2026 GMT"
|
|
dt = datetime.strptime(not_after, "%b %d %H:%M:%S %Y %Z")
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
expires_iso = dt.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
now = datetime.now(timezone.utc)
|
|
days_remaining = int((dt - now).total_seconds() // 86400)
|
|
if days_remaining < 0:
|
|
status = "expired"
|
|
elif days_remaining < 30:
|
|
status = "expiring"
|
|
else:
|
|
status = "valid"
|
|
except ValueError:
|
|
status = "unknown"
|
|
|
|
return _ok(
|
|
"parsed",
|
|
cn=cn,
|
|
not_before=not_before,
|
|
not_after=not_after,
|
|
expires_iso=expires_iso,
|
|
days_remaining=days_remaining,
|
|
status=status,
|
|
serial=serial,
|
|
sha256=sha256,
|
|
path=cert_path,
|
|
)
|
|
|
|
|
|
def list_certificates(cert_dir: str) -> tuple[bool, str, list]:
|
|
"""List certificates in ``cert_dir``. Returns plain list on success,
|
|
not the (ok, msg, data) tuple — for easier template iteration.
|
|
|
|
Returns a list of dicts with all fields; non-certificate files are skipped.
|
|
"""
|
|
ensured_ok, _, _ = _ensure_dir(cert_dir)
|
|
if not ensured_ok:
|
|
return False, "目录不可访问", []
|
|
|
|
items: list[dict] = []
|
|
try:
|
|
names = sorted(os.listdir(cert_dir))
|
|
except OSError as e:
|
|
return False, f"无法列出目录: {e}", []
|
|
|
|
for n in names:
|
|
full = os.path.join(cert_dir, n)
|
|
if not os.path.isfile(full):
|
|
continue
|
|
# only show files with allowed extensions
|
|
ext = os.path.splitext(n)[1].lower()
|
|
if ext not in ALLOWED_EXTENSIONS:
|
|
continue
|
|
st = os.stat(full)
|
|
try:
|
|
ok, msg, data = parse_pem_info(full)
|
|
except Exception:
|
|
ok, msg, data = False, "解析异常", {}
|
|
items.append({
|
|
"name": n,
|
|
"path": full,
|
|
"size": st.st_size,
|
|
"mtime": st.st_mtime,
|
|
"ext": ext,
|
|
"is_key": (ext == ".key"),
|
|
"parsed_ok": ok,
|
|
"parse_error": "" if ok else msg,
|
|
"cn": data.get("cn", ""),
|
|
"not_after": data.get("not_after", ""),
|
|
"expires_iso": data.get("expires_iso", ""),
|
|
"days_remaining": data.get("days_remaining"),
|
|
"status": data.get("status", "unknown"),
|
|
"serial": data.get("serial", ""),
|
|
"sha256": data.get("sha256", ""),
|
|
})
|
|
return True, "OK", items
|
|
|
|
|
|
def upload_certificate(
|
|
uploaded_file, cert_dir: str
|
|
) -> tuple[bool, str, dict]:
|
|
"""Accept a Werkzeug FileStorage (or any object with .filename + .read)
|
|
and save it into ``cert_dir``.
|
|
|
|
Limits: 1 MiB, allowed extensions: .pem / .crt / .key.
|
|
"""
|
|
if not uploaded_file:
|
|
return _err("未提供上传文件")
|
|
|
|
# Many test/CLI paths pass a path string or None.
|
|
filename = getattr(uploaded_file, "filename", "") or ""
|
|
name = _safe_filename(filename)
|
|
if not name:
|
|
return _err("文件名不合法 (不允许 / 或 .. 或特殊字符)")
|
|
|
|
ext = os.path.splitext(name)[1].lower()
|
|
if ext not in ALLOWED_EXTENSIONS:
|
|
return _err(
|
|
f"不支持的扩展名 '{ext}', 仅接受 {', '.join(ALLOWED_EXTENSIONS)}"
|
|
)
|
|
|
|
full, err = _safe_join(cert_dir, name)
|
|
if err or full is None:
|
|
return _err(err or "路径不安全")
|
|
|
|
# Read bytes (size cap).
|
|
try:
|
|
if hasattr(uploaded_file, "read"):
|
|
data = uploaded_file.read()
|
|
elif isinstance(uploaded_file, (bytes, bytearray)):
|
|
data = bytes(uploaded_file)
|
|
elif isinstance(uploaded_file, str) and os.path.isfile(uploaded_file):
|
|
with open(uploaded_file, "rb") as fh:
|
|
data = fh.read()
|
|
else:
|
|
return _err("无法读取上传文件内容")
|
|
except OSError as e:
|
|
return _err(f"读取失败: {e}")
|
|
|
|
if data is None:
|
|
return _err("上传内容为空")
|
|
if not isinstance(data, (bytes, bytearray)):
|
|
return _err("上传内容必须是二进制")
|
|
if len(data) == 0:
|
|
return _err("上传内容为空")
|
|
if len(data) > MAX_UPLOAD_BYTES:
|
|
return _err(
|
|
f"文件过大 ({len(data)} 字节), 上限 {MAX_UPLOAD_BYTES} 字节 (1MB)"
|
|
)
|
|
|
|
# Sanity check: PEM must contain BEGIN ... header for certs/keys.
|
|
head = data[:120].decode("ascii", errors="replace")
|
|
looks_like_pem = "-----BEGIN" in head
|
|
if ext in (".pem", ".crt") and not looks_like_pem:
|
|
return _err(
|
|
"文件内容不像有效 PEM (缺少 '-----BEGIN' 头), 上传被拒绝"
|
|
)
|
|
|
|
# Avoid overwriting existing file silently.
|
|
if os.path.exists(full):
|
|
return _err(f"目标已存在: {name}, 请先删除或重命名后再上传")
|
|
|
|
try:
|
|
with open(full, "wb") as fh:
|
|
fh.write(data)
|
|
except OSError as e:
|
|
return _err(f"保存失败: {e}")
|
|
|
|
fingerprint = hashlib.sha256(data).hexdigest().lower()
|
|
kind = "key" if ext == ".key" else (
|
|
"cert+key" if ext == ".pem" else "cert"
|
|
)
|
|
return _ok(
|
|
f"已上传 {name} ({len(data)} 字节)",
|
|
name=name,
|
|
path=full,
|
|
size=len(data),
|
|
sha256=fingerprint,
|
|
kind=kind,
|
|
ext=ext,
|
|
)
|
|
|
|
|
|
def delete_certificate(name: str, cert_dir: str) -> tuple[bool, str, dict]:
|
|
"""Delete a single file (cert or key) from ``cert_dir``."""
|
|
full, err = _safe_join(cert_dir, name)
|
|
if err or full is None:
|
|
return _err(err or "路径不安全")
|
|
if not os.path.isfile(full):
|
|
return _err(f"文件不存在: {name}")
|
|
try:
|
|
# Compute SHA-256 of content before deletion for audit trail.
|
|
with open(full, "rb") as fh:
|
|
blob = fh.read()
|
|
fingerprint = hashlib.sha256(blob).hexdigest().lower()
|
|
os.remove(full)
|
|
except OSError as e:
|
|
return _err(f"删除失败: {e}")
|
|
return _ok(
|
|
f"已删除 {name}",
|
|
name=name,
|
|
sha256=fingerprint,
|
|
size=len(blob),
|
|
)
|