admin/Nexus
1
0
Fork 0
Code Issues Pull Requests Actions 82 Packages Projects Releases Wiki Activity

feat(ops): add production probe with dual-path health checks.

Browse Source 
SSH origin checks cover auth contracts and once-schedule create/delete
when external HTTPS is blocked; includes install script for probe credentials.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Nexus Agent
2026-06-09 03:28:17 +08:00
parent 9f78207aa2
commit b62f1039db
8 changed files with 653 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker/.env.prod.example
+6
View File
@@ -29,5 +29,11 @@ NEXUS_REDIS_URL=redis://host.docker.internal:6379/0
# 宿主机 git 仓库路径（容器内安装向导用于提示 nx cron / nx update）
NEXUS_HOST_ROOT=/opt/nexus
# Optional — production probe / E2E only (host .env.prod; NOT for API runtime; never commit values)
# NEXUS_TEST_ADMIN_PASSWORD — must match admin login password
# NEXUS_PROBE_CLIENT_IP — optional; IP in login allowlist for origin probe (auto-read from DB if unset)
# NEXUS_TEST_ADMIN_PASSWORD=
# NEXUS_PROBE_CLIENT_IP=
# Optional
# NEXUS_PUBLISH_PORT=8600
docs/changelog/2026-06-09-prod-probe-health-check.md
+39
View File
@@ -0,0 +1,39 @@
# 2026-06-09 生产探针与健康检查补全
## 摘要
新增 `prod_probe.py` 双路径探针（外网 + SSH 源站），补全鉴权契约检查；提供 `prod_probe_install.sh` 在生产写入 `NEXUS_TEST_ADMIN_PASSWORD`。
## 动机
- 生产未配置探针口令，鉴权段长期 SKIP
- 国内线路外网 HTTPS 可能 RST，需源站 `127.0.0.1:8600` 探针仍可用
- 告警分页、once 调度等关键修复缺少生产自动化验收
## 涉及文件
- `scripts/prod_probe.py` — 探针主逻辑
- `scripts/prod_probe_install.sh` — 宿主机 `.env.prod` 写入探针口令
- `scripts/prod_health_check.sh` — 薄封装
- `docker/.env.prod.example` — 探针变量说明
- `tests/test_prod_probe.py`
- `docs/design/plans/2026-06-09-prod-probe-health-check.md`
- `docs/project/nexus-1panel-operations-knowledge.md`
## 迁移 / 重启
无需重启。探针口令：运维一次性执行 `prod_probe_install.sh`。
## 验证
```bash
.venv/bin/pytest tests/test_prod_probe.py -q
bash scripts/prod_health_check.sh
NEXUS_TEST_ADMIN_PASSWORD='…' bash scripts/prod_probe_install.sh && bash scripts/prod_health_check.sh
```
生产已跑通（2026-06-09）：外网 WARN（线路 RST）+ 源站鉴权全绿；once 调度 201 创建并删除。
## 注意
启用**登录 IP 白名单**时，源站探针经 `X-Real-IP` 模拟白名单 IP；可设 `NEXUS_PROBE_CLIENT_IP` 或自动读 `login_manual_ips` 首项。
docs/design/plans/2026-06-09-prod-probe-health-check.md
+42
View File
@@ -0,0 +1,42 @@
# 实施说明 — 生产探针与健康检查补全（2026-06-09）
## 背景
- `prod_health_check.sh` 鉴权段依赖 `NEXUS_TEST_ADMIN_PASSWORD`，生产 `.env.prod` 未配置且文件权限 root，导致 SKIP。
- 国内运营商线路可能阻断外网 HTTPS，但 SSH 源站 `127.0.0.1:8600` 可用。
## 方案
1. **`scripts/prod_probe.py`**：统一探针逻辑（外网公开 / SSH 源站 / 鉴权契约）。
2. **`scripts/prod_probe_install.sh`**：在宿主机 `.env.prod` 写入探针口令（与 admin 登录一致，不入 git）。
3. **`prod_health_check.sh`**：薄封装，调用 `prod_probe.py`。
## 鉴权检查项
| 端点 | 断言 |
|------|------|
| `POST /api/auth/login` | 返回 `access_token` |
| `GET /health/detail` | `checks.mysql` / `checks.redis` = ok |
| `GET /api/alert-history/stats` | today/active/recovered/top_server |
| `GET /api/alert-history/?per_page=50` | items + total + per_page=50 |
| 命令/会话分页 | items + total |
| `POST /api/schedules/` once+push | HTTP 201，随后 DELETE 清理 |
## 涉及文件
- `scripts/prod_probe.py`（新）
- `scripts/prod_probe_install.sh`（新）
- `scripts/prod_health_check.sh`
- `docker/.env.prod.example`
- `docs/project/nexus-1panel-operations-knowledge.md`（探针小节）
## 验证
```bash
bash scripts/prod_health_check.sh
NEXUS_TEST_ADMIN_PASSWORD='…' bash scripts/prod_probe_install.sh # 一次性配置生产
```
## 回滚
- 删除 `.env.prod` 中 `NEXUS_TEST_ADMIN_PASSWORD=` 行；探针鉴权段恢复 SKIP。
docs/project/nexus-1panel-operations-knowledge.md
+22 -3
View File
@@ -397,7 +397,26 @@ sudo nx verify # 已安装时应 PASS「install.html 已归档 → HTTP 404」
---
## 8. 验收清单
## 8. 生产探针（post-deploy）
部署后从开发机巡检（外网 HTTPS 可能被运营商重置时，**源站 SSH 探针仍可通过**）：
```bash
bash scripts/prod_health_check.sh # auto：外网尽力 + SSH 源站必检
NEXUS_PROBE_MODE=origin bash scripts/prod_health_check.sh # 仅源站
```
一次性在宿主机写入探针口令（**须与 admin 登录密码一致**，不入 git）：
```bash
NEXUS_TEST_ADMIN_PASSWORD='你的admin密码' bash scripts/prod_probe_install.sh
```
写入路径：`/opt/nexus/docker/.env.prod`（root 600）。鉴权段覆盖：`/health/detail`、告警 `per_page=50`、once 调度 201 创建并删除。
---
## 9. 验收清单
```bash
cd /opt/nexus && sudo nx verify
@@ -416,7 +435,7 @@ cd /opt/nexus && sudo nx verify
---
## 9. 提交修复索引
## 10. 提交修复索引
| 提交 | 主题 |
|------|------|
@@ -432,7 +451,7 @@ cd /opt/nexus && sudo nx verify
---
## 10. 与功能开发 SSOT 的差异
## 11. 与功能开发 SSOT 的差异
| 点 | `nexus-functional-development-guide.md` §16 | 1Panel 实装 |
|----|---------------------------------------------|-------------|
scripts/prod_health_check.sh
+11 -66
View File
@@ -1,5 +1,13 @@
#!/usr/bin/env bash
# Quick production health + fixed API contract checks (no secrets in output).
# Production health + API contract probe (wrapper).
# See scripts/prod_probe.py for logic.
#
# Usage:
# bash scripts/prod_health_check.sh
# NEXUS_PROBE_MODE=origin bash scripts/prod_health_check.sh # skip external HTTPS
#
# One-time prod setup (probe password = admin login password):
# NEXUS_TEST_ADMIN_PASSWORD='…' bash scripts/prod_probe_install.sh
set -euo pipefail
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
SECRETS="${ROOT}/deploy/nexus-1panel.secrets.sh"
@@ -7,68 +15,5 @@ if [[ -f "$SECRETS" ]]; then
# shellcheck disable=SC1090
source "$SECRETS"
fi
BASE="${NEXUS_PROBE_BASE:-https://api.synaglobal.vip}"
SSH_HOST="${NEXUS_SSH:-nexus}"
SSH_OPTS=(-o BatchMode=yes -o ConnectTimeout=15)
[[ -n "${NEXUS_SSH_KEY:-}" ]] && SSH_OPTS+=(-i "${NEXUS_SSH_KEY}")
echo "=== Public ==="
HEALTH=$(curl -sk --max-time 15 "${BASE}/health")
[[ "$HEALTH" == "ok" ]] || { echo "FAIL /health: ${HEALTH}"; exit 1; }
echo "✓ /health → ok"
SPA_CODE=$(curl -sk --max-time 15 -o /dev/null -w '%{http_code}' "${BASE}/app/")
[[ "$SPA_CODE" == "200" ]] || { echo "FAIL /app/: HTTP ${SPA_CODE}"; exit 1; }
echo "✓ /app/ → HTTP 200"
INSTALL_CODE=$(curl -sk --max-time 15 -o /dev/null -w '%{http_code}' "${BASE}/app/install.html")
echo " /app/install.html → HTTP ${INSTALL_CODE} (404 expected when locked)"
echo "=== Authenticated API ==="
PW="$(ssh "${SSH_OPTS[@]}" "${SSH_HOST}" \
'for f in /opt/nexus/docker/.env.prod /opt/nexus/.env /www/wwwroot/api.synaglobal.vip/.env; do
if [[ -f "$f" ]]; then
grep -E "^NEXUS_TEST_ADMIN_PASSWORD=" "$f" 2>/dev/null | cut -d= -f2- | tr -d "\""
exit 0
fi
done' \
2>/dev/null || true)"
if [[ -z "$PW" ]]; then
echo "SKIP authenticated checks (NEXUS_TEST_ADMIN_PASSWORD not found)"
exit 0
fi
TOKEN="$(curl -sk --max-time 15 -X POST "${BASE}/api/auth/login" \
-H 'Content-Type: application/json' \
-d "{\"username\":\"admin\",\"password\":\"${PW}\"}" \
| python3 -c 'import sys,json; print(json.load(sys.stdin).get("access_token",""))')"
[[ -n "$TOKEN" ]] || { echo "FAIL login"; exit 1; }
echo "✓ login OK"
TMP="$(mktemp)"
trap 'rm -f "$TMP"' EXIT
AUTH=(-H "Authorization: Bearer ${TOKEN}")
# python inline per endpoint
stats_code=$(curl -sk --max-time 15 -o "$TMP" -w '%{http_code}' "${AUTH[@]}" "${BASE}/api/alert-history/stats")
[[ "$stats_code" == "200" ]] || { echo "FAIL /api/alert-history/stats HTTP ${stats_code}"; exit 1; }
python3 -c "import json;d=json.load(open('$TMP'));assert all(k in d for k in ('today','active','recovered','top_server'))"
echo "✓ /api/alert-history/stats contract OK"
for path in \
"/api/alert-history/?page=1&per_page=5" \
"/api/assets/command-logs?page=1&per_page=5" \
"/api/assets/ssh-sessions?page=1&per_page=5"; do
code=$(curl -sk --max-time 15 -o "$TMP" -w '%{http_code}' "${AUTH[@]}" "${BASE}${path}")
[[ "$code" == "200" ]] || { echo "FAIL ${path} HTTP ${code}"; exit 1; }
python3 -c "import json;d=json.load(open('$TMP'));assert 'items' in d and 'total' in d;print(f' {path}: items={len(d[\"items\"])} total={d[\"total\"]}')"
done
echo "✓ paginated list APIs contract OK"
code=$(curl -sk --max-time 15 -o "$TMP" -w '%{http_code}' "${AUTH[@]}" "${BASE}/api/schedules/")
[[ "$code" == "200" ]] || { echo "FAIL /api/schedules/ HTTP ${code}"; exit 1; }
python3 -c "import json;d=json.load(open('$TMP'));assert isinstance(d,list);print(f' schedules count={len(d)}')"
echo "✓ /api/schedules/ OK"
echo "=== All production health checks passed ==="
MODE="${NEXUS_PROBE_MODE:-auto}"
exec python3 "${ROOT}/scripts/prod_probe.py" --mode "${MODE}"
scripts/prod_probe.py
+437
View File
@@ -0,0 +1,437 @@
#!/usr/bin/env python3
"""Production probe — public + authenticated API checks (no secrets printed).
Modes:
external — curl public BASE (may fail on some ISP routes)
origin — SSH to host, curl http://127.0.0.1:8600
auto — both; origin auth checks are required, external public is best-effort
"""
from __future__ import annotations
import argparse
import json
import os
import shlex
import subprocess
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass
from typing import Any
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@dataclass
class Config:
base_external: str
ssh_host: str
ssh_key: str | None
origin_url: str
admin_user: str
probe_password: str | None
probe_client_ip: str | None
timeout: int
def _load_secrets_env() -> None:
secrets = os.path.join(ROOT, "deploy", "nexus-1panel.secrets.sh")
if not os.path.isfile(secrets):
return
# Parse KEY=VAL / export KEY=VAL (no shell source — avoids arbitrary code)
with open(secrets, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
if line.startswith("export "):
line = line[7:].strip()
if "=" not in line:
continue
key, _, val = line.partition("=")
key = key.strip()
val = val.strip().strip("'\"")
if key and key not in os.environ:
os.environ[key] = val
def build_config() -> Config:
_load_secrets_env()
pw = os.environ.get("NEXUS_TEST_ADMIN_PASSWORD", "").strip() or None
probe_ip = os.environ.get("NEXUS_PROBE_CLIENT_IP", "").strip() or None
return Config(
base_external=os.environ.get("NEXUS_PROBE_BASE", "https://api.synaglobal.vip").rstrip("/"),
ssh_host=os.environ.get("NEXUS_SSH", "nexus"),
ssh_key=os.environ.get("NEXUS_SSH_KEY") or None,
origin_url=os.environ.get("NEXUS_PROBE_ORIGIN", "http://127.0.0.1:8600").rstrip("/"),
admin_user=os.environ.get("NEXUS_TEST_ADMIN_USER", "admin"),
probe_password=pw,
probe_client_ip=probe_ip,
timeout=int(os.environ.get("NEXUS_PROBE_TIMEOUT", "20")),
)
def ssh_base(cfg: Config) -> list[str]:
cmd = ["ssh", "-o", "BatchMode=yes", "-o", f"ConnectTimeout={cfg.timeout}"]
if cfg.ssh_key:
cmd.extend(["-i", cfg.ssh_key])
cmd.append(cfg.ssh_host)
return cmd
def fetch_env_from_origin(cfg: Config, key: str) -> str | None:
remote = (
"sudo bash -c "
+ shlex.quote(
"for f in /opt/nexus/docker/.env.prod /opt/nexus/.env "
"/www/wwwroot/api.synaglobal.vip/.env; do "
'if [[ -f "$f" ]]; then '
f'grep -E "^{key}=" "$f" 2>/dev/null | head -1 | cut -d= -f2- | tr -d \'"\'; '
"exit 0; fi; done"
)
)
try:
r = subprocess.run(
ssh_base(cfg) + [remote],
capture_output=True,
text=True,
timeout=cfg.timeout + 5,
check=False,
)
except (subprocess.TimeoutExpired, OSError):
return None
out = (r.stdout or "").strip()
return out if out else None
def fetch_password_from_origin(cfg: Config) -> str | None:
if cfg.probe_password:
return cfg.probe_password
return fetch_env_from_origin(cfg, "NEXUS_TEST_ADMIN_PASSWORD")
def fetch_probe_client_ip(cfg: Config) -> str | None:
if cfg.probe_client_ip:
return cfg.probe_client_ip
from_env = fetch_env_from_origin(cfg, "NEXUS_PROBE_CLIENT_IP")
if from_env:
return from_env
# First IP from login_manual_ips (login allowlist) for origin probe via X-Real-IP
remote = (
"sudo docker exec nexus-prod-nexus-1 python3 -c "
+ shlex.quote(
"import asyncio\n"
"from server.infrastructure.database.session import AsyncSessionLocal\n"
"from sqlalchemy import text\n"
"async def main():\n"
" async with AsyncSessionLocal() as s:\n"
" for k in ('login_manual_ips','login_allowed_ips','login_subscription_ips'):\n"
" r = await s.execute(text('SELECT value FROM settings WHERE `key` = :k'), {'k': k})\n"
" row = r.first()\n"
" if row and row[0]:\n"
" for part in str(row[0]).split(','):\n"
" ip = part.strip().split('/')[0].strip()\n"
" if ip:\n"
" print(ip); return\n"
"asyncio.run(main())\n"
)
)
try:
r = subprocess.run(
ssh_base(cfg) + [remote],
capture_output=True,
text=True,
timeout=cfg.timeout + 15,
check=False,
)
except (subprocess.TimeoutExpired, OSError):
return None
out = (r.stdout or "").strip().splitlines()
return out[0].strip() if out else None
def http_request(
url: str,
*,
method: str = "GET",
headers: dict[str, str] | None = None,
body: dict[str, Any] | None = None,
timeout: int = 20,
) -> tuple[int, str]:
data = None
req_headers = dict(headers or {})
if body is not None:
data = json.dumps(body).encode("utf-8")
req_headers.setdefault("Content-Type", "application/json")
req = urllib.request.Request(url, data=data, headers=req_headers, method=method)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.status, resp.read().decode("utf-8", errors="replace")
except urllib.error.HTTPError as e:
payload = e.read().decode("utf-8", errors="replace")
return e.code, payload
def origin_curl(
cfg: Config,
path: str,
*,
method: str = "GET",
token: str | None = None,
body: dict[str, Any] | None = None,
client_ip: str | None = None,
) -> tuple[int, str]:
url = f"{cfg.origin_url}{path}"
parts = [
"curl",
"-sS",
"--max-time",
str(cfg.timeout),
"-w",
"\\n__HTTP_CODE__%{http_code}",
"-X",
method,
]
if client_ip:
parts.extend(
["-H", f"X-Real-IP: {client_ip}", "-H", f"X-Forwarded-For: {client_ip}"]
)
if token:
parts.extend(["-H", f"Authorization: Bearer {token}"])
if body is not None:
parts.extend(["-H", "Content-Type: application/json", "-d", json.dumps(body)])
parts.append(url)
remote = " ".join(shlex.quote(p) for p in parts)
r = subprocess.run(
ssh_base(cfg) + [remote],
capture_output=True,
text=True,
timeout=cfg.timeout + 10,
check=False,
)
if r.returncode != 0 and not r.stdout:
raise RuntimeError(f"origin curl failed: {(r.stderr or '').strip() or r.returncode}")
out = r.stdout or ""
if "\n__HTTP_CODE__" not in out:
raise RuntimeError(f"origin curl bad output for {path}")
text, _, code_s = out.rpartition("\n__HTTP_CODE__")
return int(code_s), text
def check_public_external(cfg: Config) -> tuple[bool, str]:
try:
code, body = http_request(f"{cfg.base_external}/health", timeout=cfg.timeout)
if code == 200 and body.strip() == "ok":
return True, "✓ external /health → ok"
return False, f"FAIL external /health HTTP {code}: {body[:80]}"
except Exception as e:
return False, f"WARN external /health unreachable ({e}) — use origin path or VPN"
def check_public_external_spa(cfg: Config) -> tuple[bool, str]:
try:
code, _ = http_request(f"{cfg.base_external}/app/", timeout=cfg.timeout)
if code == 200:
return True, "✓ external /app/ → HTTP 200"
return False, f"FAIL external /app/ HTTP {code}"
except Exception as e:
return False, f"WARN external /app/ unreachable ({e})"
def check_public_origin(cfg: Config) -> tuple[bool, str]:
try:
code, body = origin_curl(cfg, "/health")
if code == 200 and body.strip() == "ok":
return True, "✓ origin /health → ok"
return False, f"FAIL origin /health HTTP {code}: {body[:80]}"
except Exception as e:
return False, f"FAIL origin /health ({e})"
def login_origin(cfg: Config, password: str, client_ip: str | None) -> str:
code, body = origin_curl(
cfg,
"/api/auth/login",
method="POST",
body={"username": cfg.admin_user, "password": password},
client_ip=client_ip,
)
if code != 200:
raise RuntimeError(f"login HTTP {code}: {body[:200]}")
data = json.loads(body)
token = data.get("access_token") or ""
if not token:
raise RuntimeError("login missing access_token")
return token
def assert_json(code: int, body: str, path: str) -> dict[str, Any]:
if code != 200:
raise RuntimeError(f"{path} HTTP {code}: {body[:200]}")
return json.loads(body)
def run_auth_checks(cfg: Config, token: str, client_ip: str | None) -> list[str]:
lines: list[str] = []
ip = client_ip
code, body = origin_curl(cfg, "/health/detail", token=token, client_ip=ip)
detail = assert_json(code, body, "/health/detail")
checks = detail.get("checks") or {}
for key in ("mysql", "redis"):
if checks.get(key) != "ok":
raise RuntimeError(f"/health/detail checks.{key}={checks.get(key)!r}")
lines.append("✓ /health/detail mysql+redis ok")
code, body = origin_curl(cfg, "/api/alert-history/stats", token=token, client_ip=ip)
stats = assert_json(code, body, "/api/alert-history/stats")
for key in ("today", "active", "recovered", "top_server"):
if key not in stats:
raise RuntimeError(f"alert stats missing {key}")
lines.append("✓ /api/alert-history/stats contract OK")
code, body = origin_curl(
cfg, "/api/alert-history/?page=1&per_page=50", token=token, client_ip=ip
)
alerts = assert_json(code, body, "/api/alert-history/")
if "items" not in alerts or "total" not in alerts:
raise RuntimeError("alert-history missing items/total")
if alerts.get("per_page") not in (50, None) and len(alerts["items"]) > 50:
raise RuntimeError(f"alert per_page mismatch: {alerts.get('per_page')}")
lines.append(
f"✓ alert-history per_page=50 items={len(alerts['items'])} total={alerts['total']}"
)
for path in (
"/api/assets/command-logs?page=1&per_page=5",
"/api/assets/ssh-sessions?page=1&per_page=5",
):
code, body = origin_curl(cfg, path, token=token, client_ip=ip)
data = assert_json(code, body, path)
if "items" not in data or "total" not in data:
raise RuntimeError(f"{path} missing pagination")
lines.append(f" {path}: items={len(data['items'])} total={data['total']}")
lines.append("✓ paginated command/session APIs OK")
code, body = origin_curl(cfg, "/api/schedules/", token=token, client_ip=ip)
schedules = assert_json(code, body, "/api/schedules/")
if not isinstance(schedules, list):
raise RuntimeError("/api/schedules/ not a list")
lines.append(f"✓ /api/schedules/ count={len(schedules)}")
code, body = origin_curl(cfg, "/api/servers/?per_page=1", token=token, client_ip=ip)
servers = assert_json(code, body, "/api/servers/")
items = servers.get("items") or []
if not items:
raise RuntimeError("no servers for schedule probe")
sid = items[0]["id"]
probe_name = "probe-once-health-check"
create_body = {
"name": probe_name,
"schedule_type": "push",
"run_mode": "once",
"cron_expr": "0 10 * * *",
"fire_at": "2026-12-01T10:00:00.000Z",
"source_path": "/tmp/nexus_probe",
"server_ids": f"[{sid}]",
"enabled": False,
}
code, body = origin_curl(
cfg,
"/api/schedules/",
method="POST",
token=token,
body=create_body,
client_ip=ip,
)
if code != 201:
raise RuntimeError(f"once schedule create HTTP {code}: {body[:300]}")
created = json.loads(body)
rid = created.get("id")
if not rid:
raise RuntimeError("schedule create missing id")
lines.append(f"✓ once push schedule create HTTP 201 id={rid}")
d_code, _ = origin_curl(
cfg, f"/api/schedules/{rid}", method="DELETE", token=token, client_ip=ip
)
if d_code not in (200, 204):
raise RuntimeError(f"schedule delete HTTP {d_code}")
lines.append(f"✓ probe schedule deleted id={rid}")
return lines
def main() -> int:
parser = argparse.ArgumentParser(description="Nexus production probe")
parser.add_argument(
"--mode",
choices=("auto", "external", "origin"),
default="auto",
help="auto: external best-effort + origin required",
)
args = parser.parse_args()
cfg = build_config()
print("=== Public (external) ===")
ext_public_ok = True
if args.mode in ("auto", "external"):
ok, msg = check_public_external(cfg)
print(msg)
if not ok:
ext_public_ok = False
ok2, msg2 = check_public_external_spa(cfg)
print(msg2)
if not ok2:
ext_public_ok = False
try:
code, _ = http_request(f"{cfg.base_external}/app/install.html", timeout=cfg.timeout)
print(f" /app/install.html → HTTP {code} (404 expected when locked)")
except Exception:
print(" /app/install.html → skip (external unreachable)")
else:
print(" (skipped)")
print("=== Public (origin via SSH) ===")
if args.mode in ("auto", "origin"):
ok, msg = check_public_origin(cfg)
print(msg)
if not ok:
print("=== FAILED: origin unreachable ===")
return 1
else:
print(" (skipped)")
print("=== Authenticated (origin via SSH) ===")
pw = fetch_password_from_origin(cfg)
if not pw:
print("SKIP authenticated checks (NEXUS_TEST_ADMIN_PASSWORD not on server)")
print(" → run: NEXUS_TEST_ADMIN_PASSWORD='…' bash scripts/prod_probe_install.sh")
if args.mode == "auto" and not ext_public_ok:
print("=== PASSED (origin only; external path blocked) ===")
return 0
if args.mode == "origin":
return 0
return 0
client_ip = fetch_probe_client_ip(cfg)
if not client_ip:
print("SKIP authenticated checks (no NEXUS_PROBE_CLIENT_IP / login allowlist IP)")
print(" → set NEXUS_PROBE_CLIENT_IP in .env.prod to a whitelisted login IP")
return 0
try:
token = login_origin(cfg, pw, client_ip)
print(f"✓ login OK (probe client IP {client_ip})")
for line in run_auth_checks(cfg, token, client_ip):
print(line)
except Exception as e:
print(f"FAIL authenticated: {e}")
return 1
print("=== All production probe checks passed ===")
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/prod_probe_install.sh
+59
View File
@@ -0,0 +1,59 @@
#!/usr/bin/env bash
# Install NEXUS_TEST_ADMIN_PASSWORD on production for prod_probe / E2E (not injected into API container).
#
# Usage (password must match admin login; never commit):
# NEXUS_TEST_ADMIN_PASSWORD='your-admin-password' bash scripts/prod_probe_install.sh
#
# Writes to: /opt/nexus/docker/.env.prod (root 600)
set -euo pipefail
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
SECRETS="${ROOT}/deploy/nexus-1panel.secrets.sh"
if [[ -f "$SECRETS" ]]; then
# shellcheck disable=SC1090
source "$SECRETS"
fi
PW="${NEXUS_TEST_ADMIN_PASSWORD:-}"
if [[ -z "$PW" ]]; then
echo "ERROR: set NEXUS_TEST_ADMIN_PASSWORD (same as admin login password)" >&2
exit 1
fi
if [[ ${#PW} -lt 6 ]]; then
echo "ERROR: password too short (min 6)" >&2
exit 1
fi
SSH_HOST="${NEXUS_SSH:-nexus}"
SSH_OPTS=(-o BatchMode=yes -o ConnectTimeout=20)
[[ -n "${NEXUS_SSH_KEY:-}" ]] && SSH_OPTS+=(-i "${NEXUS_SSH_KEY}")
# Pass password via env on remote (single-quoted safe payload built in Python)
B64="$(PW="$PW" python3 -c 'import base64,os; print(base64.b64encode(os.environ["PW"].encode()).decode())')"
PROBE_IP="${NEXUS_PROBE_CLIENT_IP:-}"
ssh "${SSH_OPTS[@]}" "$SSH_HOST" "sudo bash -s" <<REMOTE
set -euo pipefail
ENV="/opt/nexus/docker/.env.prod"
[[ -f "\$ENV" ]] || { echo "missing \$ENV"; exit 1; }
PW=\$(python3 -c "import base64,os; print(base64.b64decode('${B64}').decode())")
LINE="NEXUS_TEST_ADMIN_PASSWORD=\${PW}"
if grep -q '^NEXUS_TEST_ADMIN_PASSWORD=' "\$ENV"; then
sed -i "s|^NEXUS_TEST_ADMIN_PASSWORD=.*|\${LINE}|" "\$ENV"
else
printf '%s\n' "\$LINE" >> "\$ENV"
fi
if [[ -n "${PROBE_IP}" ]]; then
IPLINE="NEXUS_PROBE_CLIENT_IP=${PROBE_IP}"
if grep -q '^NEXUS_PROBE_CLIENT_IP=' "\$ENV"; then
sed -i "s|^NEXUS_PROBE_CLIENT_IP=.*|\${IPLINE}|" "\$ENV"
else
printf '%s\n' "\$IPLINE" >> "\$ENV"
fi
fi
chmod 600 "\$ENV"
grep -q '^NEXUS_TEST_ADMIN_PASSWORD=' "\$ENV"
echo "OK: NEXUS_TEST_ADMIN_PASSWORD upserted in \$ENV (value not shown)"
REMOTE
echo "Run: bash scripts/prod_health_check.sh"
tests/test_prod_probe.py
+37
View File
@@ -0,0 +1,37 @@
"""Unit tests for scripts/prod_probe.py helpers."""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
from prod_probe import assert_json, build_config # noqa: E402
def test_assert_json_ok():
data = assert_json(200, json.dumps({"a": 1}), "/x")
assert data["a"] == 1
def test_assert_json_fail_status():
with pytest.raises(RuntimeError, match="HTTP 500"):
assert_json(500, "err", "/x")
def test_build_config_defaults(monkeypatch):
monkeypatch.delenv("NEXUS_PROBE_BASE", raising=False)
monkeypatch.delenv("NEXUS_SSH", raising=False)
monkeypatch.setattr(
"prod_probe._load_secrets_env",
lambda: None,
)
cfg = build_config()
assert cfg.base_external == "https://api.synaglobal.vip"
assert cfg.ssh_host == "nexus"
assert cfg.origin_url == "http://127.0.0.1:8600"