fix: keep LXC updates running when the browser session ends

This commit is contained in:
VAIO73
2026-10-26 07:57:20 +00:00
parent af29bc9274
commit 44ab9d506d
12 changed files with 1145 additions and 40 deletions
+2
View File
@@ -13,6 +13,7 @@ import sys
import tempfile
PYTHON_FILES = (
"tests/test_lxc_update_jobs.py",
"tests/test_post_install_i18n.py",
"tests/test_remaining_cli_i18n.py",
"AppImage/scripts/tests/test_pbs_i18n.py",
@@ -33,6 +34,7 @@ PYTHON_FILES = (
"AppImage/scripts/tests/test_health_persistence_recurrence.py",
)
NODE_FILES = (
"tests/test_lxc_update_jobs.cjs",
"tests/test_api_import_boundaries.cjs",
"tests/test_api_errors.cjs",
"tests/test_optimization_labels.cjs",
+48
View File
@@ -0,0 +1,48 @@
"use client"
import { useEffect, useState } from "react"
import { Button } from "./ui/button"
import { fetchApi } from "../lib/api-config"
import { useT } from "../lib/i18n/provider"
import { viewedUpdateJobs, type LxcUpdateJob } from "../lib/lxc-update-jobs"
export function LxcUpdateJobs({ activeRunId, onResume }: {
activeRunId?: string
onResume: (job: LxcUpdateJob) => void
}) {
const t = useT()
const [jobs, setJobs] = useState<LxcUpdateJob[]>([])
const [seen, setSeen] = useState<string[]>([])
useEffect(() => {
let closed = false
const read = () => {
void fetchApi<{ jobs: LxcUpdateJob[] }>("/api/lxc-update-jobs")
.then((data) => { if (!closed) setJobs(Array.isArray(data.jobs) ? data.jobs : []) })
.catch(() => {}) // Auth recovery belongs to fetchApi, never to a job retry.
}
const viewed = () => setSeen(viewedUpdateJobs())
viewed()
read()
const timer = setInterval(read, 5000)
window.addEventListener("proxmenux-update-jobs-viewed", viewed)
return () => {
closed = true
clearInterval(timer)
window.removeEventListener("proxmenux-update-jobs-viewed", viewed)
}
}, [])
const visible = jobs.filter((job) => job.id !== activeRunId
&& (job.state !== "complete" || !seen.includes(job.id)))
if (!visible.length) return null
return <div className="space-y-2" role="status">
{visible.map((job) => <div key={job.id}
className="flex flex-wrap items-center justify-between gap-3 rounded-lg border border-violet-500/30 bg-violet-500/5 p-3">
<p className="text-sm">{t(job.state === "complete"
? "vmLxc.updates.backgroundFinished" : "vmLxc.updates.backgroundRunning",
{ vmid: job.vmid, name: job.ct_name || `CT-${job.vmid}` })}</p>
<Button variant="outline" size="sm" onClick={() => onResume(job)}>
{t("vmLxc.updates.viewBackgroundJob")}
</Button>
</div>)}
</div>
}
+67 -10
View File
@@ -54,6 +54,8 @@ interface ScriptTerminalModalProps {
completedSuccessfullyMessage?: string
completedWithErrorMessage?: (exitCode: number) => string
cancelledExitCode?: number
// Only LXC Updates opts into a separate server-owned job transport.
backgroundUpdate?: { runId: string; vmid: number; resume: boolean }
// Optional callback fired when the script's WebSocket closes
// (script_runner sends an exit code and then closes). Lets the
// parent auto-dismiss the modal — used by host-backup's Restore
@@ -72,6 +74,7 @@ export function ScriptTerminalModal({
completedSuccessfullyMessage,
completedWithErrorMessage,
cancelledExitCode,
backgroundUpdate,
onComplete,
}: ScriptTerminalModalProps) {
const t = useT()
@@ -93,6 +96,7 @@ export function ScriptTerminalModal({
const reconnectAttemptsRef = useRef(0)
const keepAliveIntervalRef = useRef<NodeJS.Timeout | null>(null)
const completionReceivedRef = useRef(false)
const outputOffsetRef = useRef(0)
const [isMobile, setIsMobile] = useState(false)
const [isTablet, setIsTablet] = useState(false)
@@ -130,7 +134,7 @@ export function ScriptTerminalModal({
}, [onComplete])
const attemptReconnect = useCallback(() => {
if (!isOpen || isComplete || reconnectAttemptsRef.current >= 3) {
if (!isOpenRef.current || isComplete || reconnectAttemptsRef.current >= 3) {
return
}
@@ -154,7 +158,9 @@ export function ScriptTerminalModal({
const wsUrl = getScriptWebSocketUrl(sessionIdRef.current)
// Single-use auth ticket appended as ?ticket=... — see lib/terminal-ws.ts.
const ws = new WebSocket(await getTicketedWsUrl(wsUrl))
const ticketedUrl = await getTicketedWsUrl(wsUrl)
if (!isOpenRef.current || !termRef.current) return
const ws = new WebSocket(ticketedUrl)
wsRef.current = ws
ws.onopen = () => {
@@ -170,9 +176,11 @@ export function ScriptTerminalModal({
}
}, 30000)
const initMessage = {
const initMessage = {
script_path: scriptPath,
params: paramsRef.current,
...(backgroundUpdate ? { resume: true, vmid: backgroundUpdate.vmid,
offset: outputOffsetRef.current } : {}),
}
ws.send(JSON.stringify(initMessage))
@@ -195,7 +203,7 @@ const initMessage = {
// The PTY worker always emits this final line. Treat it as a
// completion fallback because some WebSocket servers tear down the
// connection before the following structured message is flushed.
const exitMatch = typeof event.data === "string"
const exitMatch = !backgroundUpdate && typeof event.data === "string"
? event.data.match(/\[Script exited with code (-?\d+)\]/)
: null
if (exitMatch) {
@@ -211,6 +219,19 @@ const initMessage = {
try {
const msg = JSON.parse(event.data)
if (msg.type === "job_output") {
if (msg.truncated) termRef.current?.writeln(t("vmLxc.updates.backgroundOutputTrimmed"))
termRef.current?.write(msg.text)
outputOffsetRef.current = msg.offset
return
}
if (msg.type === "job_unavailable") {
termRef.current?.writeln(`\x1b[31m${msg.message}\x1b[0m`)
completionReceivedRef.current = true
setIsComplete(true)
ws.close()
return
}
if (msg.type === "web_interaction" && msg.interaction) {
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) {
@@ -255,6 +276,7 @@ const initMessage = {
}
ws.onclose = (event) => {
if (!isOpenRef.current) return
setConnectionStatus("offline")
if (keepAliveIntervalRef.current) {
clearInterval(keepAliveIntervalRef.current)
@@ -265,14 +287,18 @@ const initMessage = {
}
if (!isComplete && reconnectAttemptsRef.current < 3) {
reconnectTimeoutRef.current = setTimeout(attemptReconnect, 2000)
} else {
if (backgroundUpdate) {
termRef.current?.writeln(t("vmLxc.updates.backgroundDetached"))
} else {
setIsComplete(true)
onCompleteRef.current?.(-1)
}
}
}
}
}, 1000)
}, [isOpen, isComplete, scriptPath, getCompletionMessage, cancelledExitCode, t])
}, [isOpen, isComplete, scriptPath, getCompletionMessage, cancelledExitCode, t, backgroundUpdate])
const sendKey = useCallback((key: string) => {
if (!termRef.current) return
@@ -366,7 +392,9 @@ const initMessage = {
const wsUrl = getScriptWebSocketUrl(sessionIdRef.current)
// Single-use auth ticket appended as ?ticket=... — see lib/terminal-ws.ts.
const ws = new WebSocket(await getTicketedWsUrl(wsUrl))
const ticketedUrl = await getTicketedWsUrl(wsUrl)
if (!isOpenRef.current || termRef.current !== term) return
const ws = new WebSocket(ticketedUrl)
wsRef.current = ws
ws.onopen = () => {
@@ -384,6 +412,8 @@ const initMessage = {
const initMessage = {
script_path: scriptPath,
params: paramsRef.current,
...(backgroundUpdate ? { resume: backgroundUpdate.resume,
vmid: backgroundUpdate.vmid, offset: outputOffsetRef.current } : {}),
}
ws.send(JSON.stringify(initMessage))
@@ -412,7 +442,7 @@ const initMessage = {
// See the reconnect handler above. The exit line is guaranteed to be
// sent with the PTY output and is therefore a robust fallback when a
// final JSON frame is lost during server-side socket teardown.
const exitMatch = typeof event.data === "string"
const exitMatch = !backgroundUpdate && typeof event.data === "string"
? event.data.match(/\[Script exited with code (-?\d+)\]/)
: null
if (exitMatch) {
@@ -428,6 +458,19 @@ const initMessage = {
try {
const msg = JSON.parse(event.data)
if (msg.type === "job_output") {
if (msg.truncated) term.writeln(t("vmLxc.updates.backgroundOutputTrimmed"))
term.write(msg.text)
outputOffsetRef.current = msg.offset
return
}
if (msg.type === "job_unavailable") {
term.writeln(`\x1b[31m${msg.message}\x1b[0m`)
completionReceivedRef.current = true
setIsComplete(true)
ws.close()
return
}
if (msg.type === "web_interaction" && msg.interaction) {
setIsWaitingNextInteraction(false)
@@ -478,6 +521,7 @@ const initMessage = {
}
ws.onclose = (event) => {
if (!isOpenRef.current) return
setConnectionStatus("offline")
if (!completionReceivedRef.current) {
term.writeln(`\x1b[33m${t("scriptTerminal.connectionClosed")}\x1b[0m`)
@@ -489,10 +533,15 @@ const initMessage = {
}
if (!completionReceivedRef.current && !isComplete) {
if (backgroundUpdate) {
term.writeln(t("vmLxc.updates.backgroundDetached"))
attemptReconnect()
} else {
setIsComplete(true)
onCompleteRef.current?.(-1)
}
}
}
// Read `wsRef.current` inside the handler so reconnect (which swaps
// `wsRef.current` to a fresh WebSocket) doesn't leave us writing to the
@@ -578,6 +627,7 @@ const initMessage = {
sessionIdRef.current = Math.random().toString(36).substring(2, 8)
reconnectAttemptsRef.current = 0
completionReceivedRef.current = false
outputOffsetRef.current = 0
setIsComplete(false)
setInteractionInput("")
setCurrentInteraction(null)
@@ -641,8 +691,11 @@ const initMessage = {
}, [isOpen, isComplete, attemptReconnect])
const getScriptWebSocketUrl = (sid: string): string => {
const route = backgroundUpdate
? `/ws/lxc-update/${encodeURIComponent(backgroundUpdate.runId)}`
: `/ws/script/${sid}`
if (typeof window === "undefined") {
return `ws://localhost:${API_PORT}/ws/script/${sid}`
return `ws://localhost:${API_PORT}${route}`
}
const { protocol, hostname, port } = window.location
@@ -650,9 +703,9 @@ const initMessage = {
const wsProtocol = protocol === "https:" ? "wss:" : "ws:"
if (isStandardPort) {
return `${wsProtocol}//${hostname}/ws/script/${sid}`
return `${wsProtocol}//${hostname}${route}`
} else {
return `${wsProtocol}//${hostname}:${API_PORT}/ws/script/${sid}`
return `${wsProtocol}//${hostname}:${API_PORT}${route}`
}
}
@@ -664,6 +717,10 @@ const initMessage = {
if (value === "cancel" || value === "") {
setCurrentInteraction(null)
setInteractionInput("")
if (backgroundUpdate) {
wsRef.current.send(JSON.stringify({ type: "cancel" }))
return
}
handleCloseModal()
return
}
+28 -12
View File
@@ -22,6 +22,8 @@ import useSWR from "swr"
import { MetricsView } from "./metrics-dialog"
import { LxcTerminalModal } from "./lxc-terminal-modal"
import { ScriptTerminalModal } from "./script-terminal-modal"
import { LxcUpdateJobs } from "./lxc-update-jobs"
import { markUpdateJobViewed, type LxcUpdateJob } from "../lib/lxc-update-jobs"
import { OCI_CANCELLED_EXIT_CODE, shouldRefreshOciAfterCompletion } from "@/lib/script-completion"
import { LxcAppPanel, ThemeAwareLogo, ociImageLabel } from "./lxc-app-panel"
import { GuestBackupList } from "./guest-backup-list"
@@ -1998,6 +2000,7 @@ export function VirtualMachines() {
const [applyRestart, setApplyRestart] = useState(false)
const [applyStartedAt, setApplyStartedAt] = useState<number>(0)
const [applyRunId, setApplyRunId] = useState<string>("")
const [applyResume, setApplyResume] = useState(false)
const [applyCheckedOs, setApplyCheckedOs] = useState(false)
const [applyTargetIds, setApplyTargetIds] = useState<string[]>([])
const [applyTargetLabels, setApplyTargetLabels] = useState<string[]>([])
@@ -2639,6 +2642,7 @@ export function VirtualMachines() {
setApplyVmid(vmid)
setApplyTarget(target)
setApplyRunId(runId)
setApplyResume(false)
setApplyCheckedOs(opts?.checkedOs === true)
setApplyTargetIds(opts?.targetIds?.length ? opts.targetIds : fallbackTargets)
setApplyTargetLabels(opts?.targetLabels?.length ? opts.targetLabels : fallbackLabels)
@@ -2659,13 +2663,22 @@ export function VirtualMachines() {
setApplyStartedAt(Date.now())
setApplyOpen(true)
}
const handleApplyComplete = async (exitCode = 0) => {
const resumeApplyJob = (job: LxcUpdateJob) => {
setApplyVmid(job.vmid)
setApplyTarget(job.target)
setApplyRunId(job.id)
setApplyCheckedOs(job.script_path.endsWith("/install_checked_os_updates.sh"))
setApplyResume(true)
setApplyStartedAt(job.created_at * 1000)
setApplyOpen(true)
}
const handleApplyComplete = async (exitCode = 0, notifyBackend = true) => {
if (applyVmid == null) return
const duration = Math.max(0, Math.round((Date.now() - applyStartedAt) / 1000))
// The modal fires onComplete on any WS close (success or user cancel).
// Report the same run ID so the backend can collapse this compatibility
// callback with its authoritative process-completion hook.
// Legacy callers may report the same run ID to the compatibility route.
// Background Updates only refresh the view after the server's real exit.
try {
if (notifyBackend) {
await fetchApi(`/api/lxc-updates/${applyVmid}/applied`, {
method: "POST",
headers: { "Content-Type": "application/json" },
@@ -2680,6 +2693,7 @@ export function VirtualMachines() {
refresh_docker_inventory: applyDockerRefresh,
}),
})
}
} catch {
// Non-fatal — the backend process hook owns finalization.
}
@@ -7960,6 +7974,8 @@ export function VirtualMachines() {
/>
)}
<LxcUpdateJobs activeRunId={applyOpen ? applyRunId : undefined} onResume={resumeApplyJob} />
{/* LXC Update Apply Terminal — streams the apply_updates.sh
script over WS/PTY (same wiring as the hardware installers).
The backend finalizes on process exit; the close callback only
@@ -7971,7 +7987,13 @@ export function VirtualMachines() {
setApplyOpen(false)
setApplyVmid(null)
}}
onComplete={handleApplyComplete}
onComplete={(exitCode) => {
markUpdateJobViewed(applyRunId)
// The server-owned job is the sole completion authority.
void handleApplyComplete(exitCode ?? 1, false)
}}
backgroundUpdate={{ runId: applyRunId, vmid: applyVmid, resume: applyResume }}
cancelledExitCode={130}
scriptPath={applyCheckedOs
? "/usr/local/share/proxmenux/scripts/lxc/install_checked_os_updates.sh"
: "/usr/local/share/proxmenux/scripts/lxc/apply_updates.sh"}
@@ -7981,13 +8003,7 @@ export function VirtualMachines() {
completedWithErrorMessage={(exitCode) =>
t("vmLxc.updates.terminalCompletedWithError", { code: exitCode })
}
description={
applyTarget === "os"
? t("vmLxc.updates.terminalDescriptionOs")
: applyTarget === "app"
? t("vmLxc.updates.terminalDescriptionApp")
: t("vmLxc.updates.terminalDescriptionBoth")
}
description={t("vmLxc.updates.backgroundDescription")}
params={{
VMID: String(applyVmid),
RUN_ID: applyRunId,
+31
View File
@@ -0,0 +1,31 @@
/** Opaque viewer metadata only; updater commands and credentials stay server-side. */
export type LxcUpdateJob = {
id: string
vmid: number
state: "running" | "finalizing" | "complete"
target: "os" | "app" | "both"
ct_name: string
script_path: string
created_at: number
exit_code: number | null
}
const SEEN = "proxmenux-viewed-update-jobs"
export function viewedUpdateJobs(): string[] {
try {
const value = JSON.parse(localStorage.getItem(SEEN) || "[]")
return Array.isArray(value) ? value.filter((id) => typeof id === "string").slice(-128) : []
} catch {
return []
}
}
export function markUpdateJobViewed(id: string) {
try {
localStorage.setItem(SEEN, JSON.stringify([...new Set([...viewedUpdateJobs(), id])].slice(-128)))
window.dispatchEvent(new Event("proxmenux-update-jobs-viewed"))
} catch {
// Unavailable browser storage never controls or restarts server work.
}
}
+6
View File
@@ -1551,6 +1551,12 @@
"installedVersionUnverified": "Installed version is unverified. Update availability could not be confirmed.",
"versionStatusUnknown": "Update availability is unverified. Version information is unavailable or incomplete.",
"terminalTitle": "Apply updates — CT {vmid}",
"backgroundRunning": "{name} (CT {vmid}): an update is running on the server.",
"backgroundOutputTrimmed": "Earlier output was trimmed; the update continues normally.",
"backgroundFinished": "{name} (CT {vmid}): an update has finished. View its result.",
"viewBackgroundJob": "View update",
"backgroundDetached": "The viewer disconnected. The update continues on the server; reconnect after logging in.",
"backgroundDescription": "Closing this window leaves the update running. Use Ctrl+C to cancel it.",
"terminalDescriptionOs": "Applying OS package updates inside the container...",
"terminalDescriptionApp": "Running the application updater inside the container...",
"terminalDescriptionBoth": "Applying OS package and application updates inside the container...",
+6
View File
@@ -1425,6 +1425,12 @@
"installedVersionUnverified": "Nainštalovaná verzia nie je overená. Dostupnosť aktualizácie sa nepodarilo potvrdiť.",
"versionStatusUnknown": "Dostupnosť aktualizácie nie je overená. Údaje o verzii chýbajú alebo sú neúplné.",
"terminalTitle": "Aktualizácia — CT {vmid}",
"backgroundRunning": "{name} (CT {vmid}): na serveri prebieha aktualizácia.",
"backgroundOutputTrimmed": "Starší výpis bol skrátený; aktualizácia pokračuje normálne.",
"backgroundFinished": "{name} (CT {vmid}): aktualizácia sa skončila. Pozrite si výsledok.",
"viewBackgroundJob": "Zobraziť aktualizáciu",
"backgroundDetached": "Spojenie s terminálom sa prerušilo. Aktualizácia na serveri pokračuje; po prihlásení sa znova pripojte.",
"backgroundDescription": "Zatvorením okna sa aktualizácia nezastaví. Zrušiť ju môžete cez Ctrl+C.",
"terminalDescriptionOs": "Aktualizujem balíky systému v kontajneri...",
"terminalDescriptionApp": "Spúšťam aktualizáciu aplikácie v kontajneri...",
"terminalDescriptionBoth": "Aktualizujem systém aj aplikácie v kontajneri...",
+1
View File
@@ -148,6 +148,7 @@ cp "$SCRIPT_DIR/health_thresholds.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "
cp "$SCRIPT_DIR/managed_installs.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ managed_installs.py not found"
cp "$SCRIPT_DIR/lxc_update_observation.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/lxc_update_finalization.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/lxc_update_jobs.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/lxc_check_job.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/native_app_discovery.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/local_app_icons.py" "$APP_DIR/usr/bin/"
+65
View File
@@ -134,6 +134,71 @@ def _run_script_completion_hook(script_path, params, exit_code, duration_seconds
except Exception as exc:
print(f"[ProxMenux] script completion hook failed: {exc}", flush=True)
@terminal_bp.route('/api/lxc-update-jobs', methods=['GET'])
@require_admin_scope
def lxc_update_jobs_status():
"""Read-only recovery after login. This never creates or restarts a job."""
from lxc_update_jobs import list_jobs
return jsonify({'jobs': list_jobs()})
@sock.route('/ws/lxc-update/<run_id>')
def lxc_update_viewer(ws, run_id):
"""An authenticated viewer, deliberately not the owner of the update."""
if not _ws_auth_check():
ws.close(reason=1008, message="Authentication required")
return
from lxc_update_jobs import get, start
try:
init = json.loads(ws.receive(timeout=10))
if not isinstance(init, dict) or type(init.get('resume')) is not bool:
raise ValueError("Invalid update job request")
offset = init.get('offset', 0)
if type(offset) is not int or offset < 0:
raise ValueError("Invalid update output cursor")
if init['resume']:
job = get(run_id, init.get('vmid'))
else:
job = start(run_id, init.get('script_path'), init.get('params'),
_run_script_completion_hook)
revision = -1
while True:
view = job.read(offset)
if view['text'] or view['offset'] != offset:
ws.send(json.dumps({'type': 'job_output', 'text': view['text'],
'offset': view['offset'], 'truncated': view['truncated']}))
offset = view['offset']
if view['revision'] != revision:
revision = view['revision']
if view['interaction']:
ws.send(json.dumps({'type': 'web_interaction',
'interaction': view['interaction']}))
if view['state'] == 'complete':
ws.send(json.dumps({'type': 'script_complete', 'exit_code': view['exit_code']}))
break
data = ws.receive(timeout=0.2)
if data is None:
continue
try:
message = json.loads(data)
except (TypeError, ValueError):
message = data
if isinstance(message, dict) and message.get('type') == 'ping':
ws.send('{"type":"pong"}')
else:
job.input(message if isinstance(message, dict) else data)
except (ValueError, KeyError, OSError, TypeError) as exc:
try:
ws.send(json.dumps({'type': 'job_unavailable', 'message': str(exc)}))
except Exception:
pass
except Exception:
# Viewer loss, including auth expiry/reload, has no job side effects.
pass
# No process termination, PTY close or completion callback here. The job's
# worker alone owns those, even while no viewer is authenticated/connected.
@terminal_bp.route('/api/terminal/health', methods=['GET'])
def terminal_health():
"""Health check for terminal service"""
+331
View File
@@ -0,0 +1,331 @@
"""Browser-independent owners for explicitly confirmed LXC Updates runs.
Only the two Updates entrypoints are allowed. Import/list/attach never runs
anything. Interactive /ws/script consumers retain their original lifecycle.
Jobs survive viewer loss, not a Monitor/host restart; missing resume IDs fail
closed. The update shell still owns the shared scheduler/control/check flock.
"""
import codecs
import errno
import json
import os
import pty
import re
import select
import signal
import struct
import subprocess
import tempfile
import termios
import fcntl
import threading
import time
SCRIPTS = {
"/usr/local/share/proxmenux/scripts/lxc/apply_updates.sh",
"/usr/local/share/proxmenux/scripts/lxc/install_checked_os_updates.sh",
}
PARAMS = {
"VMID", "RUN_ID", "CT_NAME", "TARGET", "REQUESTED_TARGETS_JSON",
"TARGET_LABELS_JSON", "BACKUP", "BACKUP_STORAGE", "RESTART",
"UPDATE_COMMAND", "APP_NAME", "RUN_HELPER", "ALLOW_HELPER_WITH_CUSTOM",
"DOCKER_STANDALONE_TARGETS", "UPDATE_DOCKER_ENGINE",
"REFRESH_DOCKER_INVENTORY",
}
RUN_ID = re.compile(r"^[A-Za-z0-9_-]{1,128}$")
INTERACTION_ID = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
MAX_RUNTIME = 4 * 3600
RESULT_TTL = 6 * 3600
MAX_OUTPUT = 256 * 1024
MAX_JOBS = 32
_lock = threading.RLock()
_jobs = {}
class JobConflict(ValueError):
pass
def _validate(run_id, script, params):
if not isinstance(run_id, str) or not RUN_ID.fullmatch(run_id):
raise ValueError("Invalid update job ID")
if script not in SCRIPTS or os.path.realpath(script) != script:
raise ValueError("Only the LXC Updates entrypoints support background runs")
if not isinstance(params, dict) or set(params) - PARAMS:
raise ValueError("Invalid update parameters")
if any(not isinstance(v, str) or len(v) > 65536 for v in params.values()):
raise ValueError("Invalid update parameter value")
if len(json.dumps(params)) > 128 * 1024:
raise ValueError("Update parameters are too large")
if params.get("RUN_ID") != run_id:
raise ValueError("Update run ID does not match")
vmid = params.get("VMID", "")
if not re.fullmatch(r"[1-9][0-9]{0,8}", vmid):
raise ValueError("Invalid container ID")
if params.get("TARGET") not in ("os", "app", "both"):
raise ValueError("Invalid update target")
return int(vmid)
class UpdateJob:
def __init__(self, run_id, script, params, callback):
self.id, self.script, self.params = run_id, script, dict(params)
self.vmid = int(params["VMID"])
self.created_at = time.time()
self.started = time.monotonic()
self.finished_at = None
self.exit_code = None
self.state = "running"
self.output = ""
self.offset = 0
self.interaction = None
self.revision = 0
self.lock = threading.RLock()
self.callback = callback
self.stop_reason = None
self.stop_at = None
self.master = None
self.process = None
self.log_path = None
self._interaction_buffer = ""
def launch(self):
master, slave = pty.openpty()
self.master = master
try:
fd, self.log_path = tempfile.mkstemp(prefix="proxmenux_update_", suffix=".log")
os.close(fd)
fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", 30, 120, 0, 0))
env = dict(os.environ, **self.params, EXECUTION_MODE="web",
WEB_LOG=self.log_path, TERM="xterm-256color", PYTHONUNBUFFERED="1")
self.process = subprocess.Popen(
["/bin/bash", self.script], stdin=slave, stdout=slave, stderr=slave,
start_new_session=True, env=env,
)
except BaseException:
os.close(master)
self.master = None
if self.log_path:
os.unlink(self.log_path)
raise
finally:
os.close(slave)
try:
threading.Thread(target=self._run, daemon=True,
name=f"lxc-update-{self.id}").start()
except BaseException:
# A failed worker start must not leave an untracked updater alive.
try:
os.killpg(self.process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
self.process.wait()
os.close(master)
self.master = None
os.unlink(self.log_path)
raise
def _append(self, text):
with self.lock:
self.output = (self.output + text)[-MAX_OUTPUT:]
self.offset += len(text)
def _interactions(self, stream):
self._interaction_buffer += stream.read(16384)
if len(self._interaction_buffer) > 65536:
self._interaction_buffer = ""
return
lines = self._interaction_buffer.split("\n")
self._interaction_buffer = lines.pop()
for line in lines:
if not line.startswith("WEB_INTERACTION:"):
continue
try:
import base64
parts = line.rstrip("\n")[16:].split(":", 4)
kind, identity, title, message = parts[:4]
if not INTERACTION_ID.fullmatch(identity):
continue
value = dict(type=kind, id=identity,
title=base64.b64decode(title).decode(),
message=base64.b64decode(message).decode())
if len(parts) > 4:
if kind == "menu":
value["options"] = json.loads(parts[4])
elif kind in ("input", "inputbox"):
value["default"] = base64.b64decode(parts[4]).decode()
with self.lock:
self.interaction = value
self.revision += 1
except (ValueError, UnicodeError, IndexError):
continue
def _signal(self, reason):
with self.lock:
if self.process.poll() is not None or self.stop_reason:
return
self.stop_reason, self.stop_at = reason, time.monotonic()
try:
os.killpg(self.process.pid, signal.SIGTERM)
except ProcessLookupError:
pass
def cancel(self):
self._signal("cancelled")
def _run(self):
decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
eof, exited_at = False, None
try:
with open(self.log_path, encoding="utf-8", errors="replace") as interactions:
while True:
now = time.monotonic()
running = self.process.poll() is None
if not running:
exited_at = exited_at or now
if eof or now - exited_at >= 1:
break
if time.monotonic() - self.started >= MAX_RUNTIME:
self._signal("timeout")
if running and self.stop_at and now - self.stop_at >= 2:
try:
os.killpg(self.process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
self._interactions(interactions)
readable, _, _ = select.select([] if eof else [self.master], [], [], 0.05)
if readable:
try:
data = os.read(self.master, 16384)
except OSError as exc:
if exc.errno != errno.EIO:
raise
data = b""
if data:
self._append(decoder.decode(data))
continue
eof = True
code = self.process.wait()
self._append(decoder.decode(b"", final=True))
if self.stop_reason:
code = 124 if self.stop_reason == "timeout" else 130
except BaseException as exc:
self._append(f"\r\nUpdate viewer failed: {type(exc).__name__}\r\n")
self._signal("viewer failure")
try:
self.process.wait(timeout=2)
except subprocess.TimeoutExpired:
os.killpg(self.process.pid, signal.SIGKILL)
self.process.wait()
code = 1
finally:
os.close(self.master)
self.master = None
try:
os.unlink(self.log_path)
except FileNotFoundError:
pass
with self.lock:
self.exit_code = code
self.state = "finalizing"
self.interaction = None
self.revision += 1
try:
self.callback(self.script, self.params, code, time.monotonic() - self.started)
finally:
with self.lock:
self.state = "complete"
self.finished_at = time.time()
self.revision += 1
def view(self):
with self.lock:
return dict(id=self.id, vmid=self.vmid, state=self.state,
target=self.params["TARGET"], ct_name=self.params.get("CT_NAME", ""),
created_at=self.created_at, finished_at=self.finished_at,
exit_code=self.exit_code, script_path=self.script)
def read(self, offset):
with self.lock:
begin = self.offset - len(self.output)
cursor = max(begin, min(offset, self.offset))
return dict(text=self.output[cursor - begin:], offset=self.offset,
truncated=offset < begin, interaction=self.interaction,
revision=self.revision, state=self.state, exit_code=self.exit_code)
def input(self, message):
with self.lock:
if self.state != "running" or self.master is None:
return
if isinstance(message, str):
if len(message) > 16384:
raise ValueError("Terminal input is too large")
if message == "\x03":
self.cancel()
else:
os.write(self.master, message.encode())
elif message.get("type") == "cancel":
self.cancel()
elif message.get("type") == "resize":
cols = max(20, min(400, int(message.get("cols", 120))))
rows = max(5, min(200, int(message.get("rows", 30))))
fcntl.ioctl(self.master, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
elif message.get("type") == "interaction_response":
identity, value = message.get("id"), message.get("value")
if (not self.interaction or identity != self.interaction["id"]
or not isinstance(value, str) or len(value) > 16384):
raise ValueError("No matching pending interaction")
path = f"/tmp/proxmenux_response_{identity}"
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
with os.fdopen(fd, "w") as response:
response.write(value)
self.interaction = None
self.revision += 1
def start(run_id, script, params, callback):
vmid = _validate(run_id, script, params)
with _lock:
existing = _jobs.get(run_id)
if existing:
if existing.script != script or existing.params != params:
raise JobConflict("Update job ID already belongs to another request")
return existing
_prune()
if any(j.vmid == vmid and j.state != "complete" for j in _jobs.values()):
raise JobConflict("An update is already running for this container")
if len(_jobs) >= MAX_JOBS:
completed = [j for j in _jobs.values() if j.state == "complete"]
if not completed:
raise JobConflict("Too many active update jobs; try again later")
oldest = min(completed, key=lambda j: j.finished_at)
del _jobs[oldest.id]
job = UpdateJob(run_id, script, params, callback)
_jobs[run_id] = job
try:
job.launch()
except BaseException:
_jobs.pop(run_id, None)
raise
return job
def _prune():
now = time.time()
for identity, job in list(_jobs.items()):
if job.finished_at and now - job.finished_at > RESULT_TTL:
del _jobs[identity]
def get(run_id, vmid):
with _lock:
job = _jobs.get(run_id)
if job is None or job.vmid != int(vmid):
raise KeyError("Update job is unavailable; it was not restarted")
return job
def list_jobs():
with _lock:
_prune()
return [j.view() for j in sorted(_jobs.values(), key=lambda j: j.created_at, reverse=True)]
+82
View File
@@ -0,0 +1,82 @@
// Execute the actual modal handlers and Updates recovery wiring.
const assert = require('node:assert/strict')
const fs = require('node:fs'), path = require('node:path'), vm = require('node:vm')
const {loadTypeScript} = require('./ts-fixture.cjs')
const root = path.resolve(__dirname, '../AppImage')
const ts = require(path.join(root, 'node_modules/typescript'))
const source = fs.readFileSync(path.join(root, 'components/script-terminal-modal.tsx'), 'utf8')
const tree = ts.createSourceFile('modal.tsx', source, ts.ScriptTarget.Latest, true, ts.ScriptKind.TSX)
const handlers = {onmessage: [], onclose: []}
function walk(node) {
if (ts.isBinaryExpression(node) && /^ws\.(onmessage|onclose)$/.test(node.left.getText(tree)))
handlers[node.left.name.text].push(node.right.getText(tree))
ts.forEachChild(node, walk)
}
walk(tree)
assert.equal(handlers.onmessage.length, 2)
assert.equal(handlers.onclose.length, 2)
const completion = loadTypeScript('lib/script-completion.ts')
function fixture(backgroundUpdate) {
const output = [], finished = [], reconnects = []
const term = {write: s => output.push(s), writeln: s => output.push(s)}
const ws = {readyState: 1, close() { this.readyState = 3 }}
const context = {
...completion, backgroundUpdate, term, termRef: {current: term}, ws, WebSocket: {OPEN: 1},
completionReceivedRef: {current: false}, outputOffsetRef: {current: 0},
isOpenRef: {current: true}, isComplete: false, cancelledExitCode: 130,
waitingTimeoutRef: {current: null}, keepAliveIntervalRef: {current: null},
reconnectAttemptsRef: {current: 0}, reconnectTimeoutRef: {current: null},
onCompleteRef: {current: code => finished.push(code)},
setIsComplete: value => context.isComplete = value, setConnectionStatus() {},
setIsWaitingNextInteraction() {}, setCurrentInteraction: value => context.interaction = value,
clearTimeout() {}, clearInterval() {}, setTimeout: fn => { fn(); return 1 },
attemptReconnect: () => reconnects.push(true), t: key => key,
getCompletionMessage: code => 'Result ' + code,
}
return {context, output, finished, reconnects, ws}
}
function compile(code, context) {
return vm.runInNewContext(ts.transpileModule('(' + code + ')', {compilerOptions: {
target: ts.ScriptTarget.ES2020, module: ts.ModuleKind.CommonJS,
}}).outputText, context)
}
for (const handler of handlers.onmessage) {
const f = fixture({runId: 'existing', vmid: 100, resume: true})
const message = compile(handler, f.context)
message({data: JSON.stringify({type: 'job_output', text: '[Script exited with code 0]', offset: 27})})
assert.equal(f.finished.length, 0, 'text inside a job is never its authoritative result')
assert.equal(f.context.outputOffsetRef.current, 27)
message({data: JSON.stringify({type: 'web_interaction', interaction: {id:'pending', type:'yesno'}})})
assert.equal(f.context.interaction.id, 'pending', 'replayed interaction remains answerable')
message({data: JSON.stringify({type: 'script_complete', exit_code: 3})})
assert.deepEqual(f.finished, [3])
message({data: JSON.stringify({type: 'script_complete', exit_code: 0})})
assert.deepEqual(f.finished, [3], 'replayed completion cannot finalize twice')
const missing = fixture({runId: 'missing'})
compile(handler, missing.context)({data: JSON.stringify({type:'job_unavailable', message:'Unavailable; not restarted'})})
assert.equal(missing.finished.length, 0, 'missing job is not success or a synthetic failure callback')
const legacy = fixture(undefined)
compile(handler, legacy.context)({data:'[Script exited with code 0]'})
assert.deepEqual(legacy.finished, [0], 'legacy terminal exit fallback still works')
}
for (const handler of handlers.onclose) {
const f = fixture({runId: 'existing'})
compile(handler, f.context)({})
assert.equal(f.finished.length, 0, 'losing a viewer never reports update failure')
assert.equal(f.context.isComplete, false)
assert.equal(f.reconnects.length, 1)
const closed = fixture({runId:'existing'}); closed.context.isOpenRef.current = false
compile(handler, closed.context)({})
assert.equal(closed.reconnects.length, 0, 'closing the view cannot reopen a socket')
}
const local = new Map(), events = []
const storage = {getItem:k=>local.get(k), setItem:(k,v)=>local.set(k,v)}
const jobs = loadTypeScript('lib/lxc-update-jobs.ts', {localStorage:storage,
window:{dispatchEvent:e=>events.push(e)}, Event:class {constructor(type){this.type=type}}})
local.set('proxmenux-viewed-update-jobs', '{')
assert.equal(jobs.viewedUpdateJobs().length, 0)
jobs.markUpdateJobViewed('opaque-run'); jobs.markUpdateJobViewed('opaque-run')
assert.equal(jobs.viewedUpdateJobs().length, 1)
assert.equal(events.length, 2)
assert.equal([...local.values()].join(''), '["opaque-run"]', 'storage has IDs, no commands or credentials')
console.log('PASS actual initial/reconnect handlers, auth-independent recovery metadata and legacy completion')
+460
View File
@@ -0,0 +1,460 @@
"""Real inert processes and actual Flask/ticket/socket handler boundaries."""
import ast
import base64
import io
import importlib.util
import json
import os
import pty
import re
import select
import fcntl
import struct
import termios
from pathlib import Path
import shlex
import subprocess
import sys
import tempfile
import threading
import time
from types import ModuleType, SimpleNamespace
import unittest
from unittest.mock import Mock, patch
from flask import Blueprint, Flask, jsonify, request
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "AppImage/scripts"))
import lxc_update_jobs as jobs
def wait_for(predicate, timeout=6):
deadline = time.monotonic() + timeout
while not predicate():
if time.monotonic() >= deadline:
raise AssertionError("Timed out waiting for fixture")
time.sleep(0.01)
class JobTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.folder = Path(self.temp.name)
self.script = self.folder / "apply.sh"
self.gate = self.folder / "backup-done"
self.events = self.folder / "events"
self.script.write_text(
f"echo backup-start; echo backup >> {shlex.quote(str(self.events))}\n"
f"while [ ! -e {shlex.quote(str(self.gate))} ]; do sleep .02; done\n"
f"echo update-finished; echo update >> {shlex.quote(str(self.events))}\n"
)
self.allowed = patch.object(jobs, "SCRIPTS", {str(self.script)})
self.allowed.start()
self.addCleanup(self.allowed.stop)
jobs._jobs.clear()
self.callback = Mock()
self.addCleanup(self.stop_jobs)
def stop_jobs(self):
for job in list(jobs._jobs.values()):
if job.state != "complete":
job.cancel()
wait_for(lambda: job.state == "complete")
jobs._jobs.clear()
def params(self, run="run-1", vmid="100"):
return dict(VMID=vmid, RUN_ID=run, TARGET="app", CT_NAME="fixture", BACKUP="1")
def start(self, run="run-1", vmid="100"):
return jobs.start(run, str(self.script), self.params(run, vmid), self.callback)
def finish(self, job):
self.gate.touch()
wait_for(lambda: job.state == "complete")
def test_real_backup_and_update_continue_without_any_viewer(self):
job = self.start()
wait_for(lambda: self.events.exists())
self.assertEqual(self.events.read_text().splitlines(), ["backup"])
self.assertEqual(job.state, "running")
self.finish(job)
self.assertEqual(self.events.read_text().splitlines(), ["backup", "update"])
self.assertEqual(job.exit_code, 0)
self.callback.assert_called_once()
self.assertEqual(self.callback.call_args.args[:3],
(str(self.script), self.params(), 0))
self.assertIn("update-finished", job.read(0)["text"])
def test_same_run_reconnect_is_idempotent_and_other_ct_cannot_attach(self):
job = self.start()
self.assertIs(self.start(), job)
with self.assertRaises(KeyError):
jobs.get(job.id, 200)
with self.assertRaises(KeyError):
jobs.get("lost-run", 100)
self.finish(job)
self.assertIs(jobs.get(job.id, 100), job)
self.assertIs(self.start(), job)
self.assertEqual(self.events.read_text().splitlines(), ["backup", "update"])
self.callback.assert_called_once()
def test_different_run_for_same_ct_rejected_before_second_spawn(self):
job = self.start()
with patch.object(jobs.subprocess, "Popen") as spawn:
with self.assertRaises(jobs.JobConflict):
self.start("run-2")
spawn.assert_not_called()
self.finish(job)
other = self.start("run-2")
wait_for(lambda: other.state == "complete")
self.assertEqual(self.events.read_text().splitlines().count("update"), 2)
def test_concurrent_same_id_starts_once(self):
calls, errors = [], []
barrier = threading.Barrier(6)
def attempt():
barrier.wait()
try:
calls.append(self.start())
except Exception as exc:
errors.append(exc)
threads = [threading.Thread(target=attempt) for _ in range(6)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
self.assertFalse(errors)
self.assertEqual(len({id(job) for job in calls}), 1)
self.finish(calls[0])
self.assertEqual(self.events.read_text().splitlines(), ["backup", "update"])
def test_invalid_request_has_zero_process_and_zero_tempfile_effects(self):
bad = [
("../bad", str(self.script), self.params()),
("run-1", "/bin/bash", self.params()),
("run-1", str(self.script), dict(self.params(), LD_PRELOAD="/bad")),
("run-1", str(self.script), dict(self.params(), VMID="0")),
("run-1", str(self.script), dict(self.params(), RUN_ID="other")),
("run-1", str(self.script), dict(self.params(), UPDATE_COMMAND=123)),
("run-1", str(self.script), dict(self.params(), TARGET="unexpected")),
]
with patch.object(jobs.subprocess, "Popen") as spawn, \
patch.object(jobs.tempfile, "mkstemp") as files:
for identity, script, params in bad:
with self.subTest(params=params), self.assertRaises(ValueError):
jobs.start(identity, script, params, self.callback)
spawn.assert_not_called()
files.assert_not_called()
self.assertFalse(jobs.list_jobs())
def test_request_id_cannot_be_rebound_to_another_command(self):
job = self.start()
with patch.object(jobs.subprocess, "Popen") as spawn:
with self.assertRaises(jobs.JobConflict):
jobs.start("run-1", str(self.script),
dict(self.params(), UPDATE_COMMAND="different"), self.callback)
spawn.assert_not_called()
self.finish(job)
def test_failure_and_launch_error_are_not_reported_as_success(self):
self.script.write_text("echo failed-backup; exit 3\n")
job = self.start()
wait_for(lambda: job.state == "complete")
self.assertEqual(job.exit_code, 3)
self.assertFalse(self.events.exists())
self.callback.assert_called_once()
with patch.object(jobs.subprocess, "Popen", side_effect=OSError("cannot execute")):
with self.assertRaises(OSError):
self.start("run-2")
self.assertNotIn("run-2", jobs._jobs)
def test_cancel_is_explicit_and_kills_the_inert_process_group(self):
job = self.start()
wait_for(lambda: self.events.exists())
job.input({"type": "cancel"})
wait_for(lambda: job.state == "complete")
self.assertEqual(job.exit_code, 130)
self.assertEqual(self.events.read_text().splitlines(), ["backup"])
self.callback.assert_called_once()
def test_timeout_is_bounded_even_if_the_child_closes_its_terminal(self):
self.script.write_text("exec </dev/null >/dev/null 2>&1\nwhile :; do sleep .02; done\n")
with patch.object(jobs, "MAX_RUNTIME", .15):
job = self.start()
wait_for(lambda: job.state == "complete")
self.assertEqual(job.exit_code, 124)
self.callback.assert_called_once()
def test_output_is_bounded_and_cursor_replay_is_read_only(self):
job = self.start()
wait_for(lambda: self.events.exists())
with patch.object(jobs, "MAX_OUTPUT", 32):
job._append("x" * 100)
snap = job.read(0)
self.assertEqual(len(snap["text"]), 32)
self.assertTrue(snap["truncated"])
self.assertEqual(job.read(snap["offset"])["text"], "")
self.finish(job)
def test_unknown_interaction_response_never_writes_a_file(self):
job = self.start()
with patch.object(jobs.os, "open") as files:
with self.assertRaises(ValueError):
job.input({"type": "interaction_response", "id": "not-pending", "value": "yes"})
files.assert_not_called()
self.finish(job)
def test_real_process_exit_calls_existing_backend_finalizer_without_viewer(self):
source = ast.parse((ROOT / "AppImage/scripts/flask_server.py").read_text())
node = next(n for n in source.body if isinstance(n, ast.FunctionDef)
and n.name == "_terminal_lxc_update_completed")
finalize = Mock()
scope = dict(os=os, _LXC_APPLY_UPDATES_SCRIPT=str(self.script),
_LXC_INSTALL_CHECKED_OS_SCRIPT=str(self.folder/'checked.sh'),
_normalise_lxc_update_run_id=lambda value, **_: value,
_normalise_lxc_update_targets=lambda value, target: value or [target],
_json_list=lambda value: json.loads(value or '[]'),
_finalize_lxc_update=finalize)
exec(compile(ast.Module(body=[node], type_ignores=[]), 'actual-completion-hook', 'exec'), scope)
def callback(script, params, code, duration):
scope['_terminal_lxc_update_completed'](script_path=script, params=params,
exit_code=code, duration_seconds=duration)
self.callback.side_effect = callback
job = self.start()
self.finish(job)
finalize.assert_called_once()
args, kwargs = finalize.call_args
self.assertEqual(args, (100,))
self.assertEqual(kwargs['status'], 'success')
self.assertEqual(kwargs['run_id'], 'run-1')
self.assertEqual(kwargs['target'], 'app')
self.assertEqual(kwargs['requested_targets'], ['app'])
def test_list_never_starts_or_probes_a_guest_and_retention_has_safe_margin(self):
job = self.start()
with patch.object(jobs.subprocess, "Popen") as spawn:
for _ in range(5):
self.assertEqual(jobs.list_jobs()[0]["id"], job.id)
spawn.assert_not_called()
self.assertGreater(jobs.RESULT_TTL, jobs.MAX_RUNTIME)
self.finish(job)
ended = job.finished_at
with patch.object(jobs.time, "time", return_value=ended + jobs.RESULT_TTL):
self.assertEqual(len(jobs.list_jobs()), 1)
with patch.object(jobs.time, "time", return_value=ended + jobs.RESULT_TTL + .01):
self.assertEqual(jobs.list_jobs(), [])
def test_full_history_evicts_only_completed_jobs_not_running_work(self):
first = self.start()
self.finish(first)
with patch.object(jobs, "MAX_JOBS", 1):
second = self.start("run-2")
self.assertNotIn(first.id, jobs._jobs)
wait_for(lambda: second.state == "complete")
self.assertEqual(self.callback.call_count, 2)
def test_tempfile_and_worker_start_failures_leave_no_job_or_process(self):
with patch.object(jobs.tempfile, "mkstemp", side_effect=OSError("disk full")), \
patch.object(jobs.subprocess, "Popen") as spawn:
with self.assertRaises(OSError):
self.start()
spawn.assert_not_called()
actual_spawn = jobs.subprocess.Popen
children = []
def launch(*args, **kwargs):
child = actual_spawn(*args, **kwargs)
children.append(child)
return child
with patch.object(jobs.subprocess, "Popen", side_effect=launch), \
patch.object(jobs.threading.Thread, "start", side_effect=RuntimeError("no worker")):
with self.assertRaises(RuntimeError):
self.start()
self.assertIsNotNone(children[0].poll())
self.assertEqual(jobs.list_jobs(), [])
self.callback.assert_not_called()
def test_pending_interaction_survives_partial_log_and_reconnect(self):
job = self.start()
identity = "fixture-" + os.urandom(8).hex()
response = Path("/tmp/proxmenux_response_" + identity)
self.addCleanup(lambda: response.unlink(missing_ok=True))
encode = lambda s: base64.b64encode(s.encode()).decode()
line = f"WEB_INTERACTION:inputbox:{identity}:{encode('Title')}:{encode('Question')}:{encode('default')}\n"
job._interactions(io.StringIO(line[:20]))
self.assertIsNone(job.read(0)["interaction"])
job._interactions(io.StringIO(line[20:]))
self.assertEqual(jobs.get(job.id, 100).read(0)["interaction"]["default"], "default")
job.input({"type": "interaction_response", "id": identity, "value": "answer"})
self.assertEqual(response.read_text(), "answer")
self.assertEqual(response.stat().st_mode & 0o777, 0o600)
with patch.object(jobs.os, "open") as files:
with self.assertRaises(ValueError):
job.input({"type": "interaction_response", "id": identity, "value": "again"})
files.assert_not_called()
self.finish(job)
class RouteTests(unittest.TestCase):
stop_jobs = JobTests.stop_jobs
params = JobTests.params
finish = JobTests.finish
def setUp(self):
JobTests.setUp(self)
self.app = Flask(__name__)
self.auth = ModuleType("auth_manager")
self.auth.load_auth_config = lambda: {"enabled": True}
self.auth.verify_token = lambda token: "fixture" if token == "valid" else None
self.auth.verify_token_full = lambda token: (
("fixture", "full_admin") if token == "valid" else
("fixture", "read_only") if token == "read-only" else (None, None))
self.auth_patch = patch.dict(sys.modules, {"auth_manager": self.auth})
self.auth_patch.start()
self.addCleanup(self.auth_patch.stop)
spec = importlib.util.spec_from_file_location(
"update_jobs_jwt_fixture", ROOT / "AppImage/scripts/jwt_middleware.py")
middleware = importlib.util.module_from_spec(spec)
spec.loader.exec_module(middleware)
self.ws_routes = {}
def route(path):
def register(fn):
self.ws_routes[path] = fn
return fn
return register
scope = dict(terminal_bp=Blueprint("jobs_fixture", __name__),
sock=SimpleNamespace(route=route), request=request, jsonify=jsonify,
require_admin_scope=middleware.require_admin_scope,
_TERMINAL_TICKETS={}, _TICKETS_LOCK=threading.Lock(),
_TICKET_TTL=5, _TICKET_MAX_INFLIGHT=256,
time=time, secrets=__import__("secrets"), json=json,
_run_script_completion_hook=self.callback)
names = {"_issue_terminal_ticket", "_consume_terminal_ticket", "_ws_auth_check",
"issue_terminal_ticket_route", "lxc_update_jobs_status", "lxc_update_viewer"}
nodes = [n for n in ast.parse((ROOT / "AppImage/scripts/flask_terminal_routes.py").read_text()).body
if isinstance(n, ast.FunctionDef) and n.name in names]
self.assertEqual(len(nodes), len(names))
exec(compile(ast.Module(body=nodes, type_ignores=[]), "actual-terminal-routes", "exec"), scope)
self.scope = scope
self.app.register_blueprint(scope["terminal_bp"])
self.client = self.app.test_client()
def ticket(self, token="valid"):
return self.client.post("/api/terminal/ticket",
headers={"Authorization": "Bearer " + token})
def attach(self, init, ticket=None, messages=None):
frames = iter([json.dumps(init), *(messages or [])])
ws = Mock()
def receive(**_):
try:
return next(frames)
except StopIteration:
raise ConnectionError("browser reload / connection closed")
ws.receive.side_effect = receive
path = "/ws/lxc-update/run-1" + ("?ticket=" + ticket if ticket else "")
with self.app.test_request_context(path):
self.ws_routes["/ws/lxc-update/<run_id>"](ws, "run-1")
return ws
def test_expired_login_cannot_start_and_has_zero_job_effects(self):
for token, status in [("expired", 401), ("read-only", 403)]:
with self.subTest(token=token), patch.object(jobs.subprocess, "Popen") as launch:
response = self.ticket(token)
self.assertEqual(response.status_code, status)
self.attach(dict(resume=False, script_path=str(self.script), params=self.params()))
launch.assert_not_called()
self.assertFalse(jobs.list_jobs())
def test_expiry_during_backup_does_not_cancel_or_restart_and_fresh_login_resumes(self):
ticket = self.ticket().json["ticket"]
self.attach(dict(resume=False, script_path=str(self.script), params=self.params()), ticket)
job = jobs.get("run-1", 100)
wait_for(lambda: self.events.exists())
denied = self.client.get("/api/lxc-update-jobs",
headers={"Authorization": "Bearer expired"})
self.assertEqual(denied.status_code, 401)
self.assertIsNone(job.stop_reason)
self.assertEqual(self.client.get("/api/lxc-update-jobs",
headers={"Authorization": "Bearer valid"}).json["jobs"][0]["id"], job.id)
with patch.object(jobs.subprocess, "Popen") as launch:
self.attach(dict(resume=True, vmid=100), self.ticket().json["ticket"])
launch.assert_not_called()
self.finish(job)
self.assertEqual(job.exit_code, 0)
self.assertEqual(self.events.read_text().splitlines(), ["backup", "update"])
self.callback.assert_called_once()
complete = self.attach(dict(resume=True, vmid=100), self.ticket().json["ticket"])
frames = [json.loads(c.args[0]) for c in complete.send.call_args_list]
self.assertEqual(frames[-1], {"type": "script_complete", "exit_code": 0})
def test_ticket_is_single_use_and_invalid_resume_never_becomes_execution(self):
ticket = self.ticket().json["ticket"]
with patch.object(jobs.subprocess, "Popen") as launch:
ws = self.attach(dict(resume=True, vmid=100), ticket)
self.assertIn("job_unavailable", ws.send.call_args.args[0])
denied = self.attach(dict(resume=True, vmid=100), ticket)
denied.close.assert_called_once()
launch.assert_not_called()
def test_expiry_during_update_then_two_viewers_cannot_duplicate_or_finalize_early(self):
update_gate = self.folder / "update-done"
self.script.write_text(self.script.read_text() +
f"while [ ! -e {shlex.quote(str(update_gate))} ]; do sleep .02; done\n")
self.attach(dict(resume=False, script_path=str(self.script), params=self.params()),
self.ticket().json["ticket"])
job = jobs.get("run-1", 100)
self.gate.touch()
wait_for(lambda: self.events.exists() and "update" in self.events.read_text())
for token, status in [("expired", 401), ("read-only", 403)]:
self.assertEqual(self.client.get("/api/lxc-update-jobs", headers={
"Authorization": "Bearer " + token}).status_code, status)
self.callback.assert_not_called()
with patch.object(jobs.subprocess, "Popen") as spawn:
for _ in range(2):
self.attach(dict(resume=True, vmid=100), self.ticket().json["ticket"])
spawn.assert_not_called()
self.assertIsNone(job.stop_reason)
update_gate.touch()
wait_for(lambda: job.state == "complete")
self.callback.assert_called_once()
self.assertEqual(self.events.read_text().splitlines(), ["backup", "update"])
def test_original_interactive_route_still_terminates_its_process_on_disconnect(self):
# Reproduce the original ownership bug using the untouched legacy route.
# Only its allowed fixture directory and process observation are injected.
nodes = [n for n in ast.parse((ROOT / "AppImage/scripts/flask_terminal_routes.py").read_text()).body
if isinstance(n, ast.FunctionDef) and n.name == "script_websocket"]
realpath = os.path.realpath
os_view = SimpleNamespace(**vars(os))
os_view.path = SimpleNamespace(**vars(os.path))
os_view.path.realpath = lambda p: (str(self.folder)
if p == "/usr/local/share/proxmenux/scripts" else realpath(p))
children = []
def spawn(*args, **kwargs):
child = subprocess.Popen(*args, **kwargs)
children.append(child)
self.addCleanup(lambda: child.poll() is None and child.kill())
return child
self.scope.update(os=os_view, subprocess=SimpleNamespace(Popen=spawn),
tempfile=tempfile, pty=pty, fcntl=fcntl, select=select, base64=base64, threading=threading,
_SAFE_ID_RE=re.compile(r"^[A-Za-z0-9_-]{1,64}$"),
set_winsize=lambda *_: None)
exec(compile(ast.Module(body=nodes, type_ignores=[]), 'actual-legacy-route', 'exec'), self.scope)
ws = Mock()
def receive(**_):
if not children:
return json.dumps(dict(script_path=str(self.script), params=self.params()))
wait_for(lambda: self.events.exists())
raise ConnectionError("expired login reloaded browser")
ws.receive.side_effect = receive
with self.app.test_request_context('/ws/script/fixture?ticket='+self.ticket().json['ticket']):
self.ws_routes['/ws/script/<session_id>'](ws, 'fixture')
self.assertIsNotNone(children[0].poll())
self.assertNotEqual(children[0].returncode, 0)
self.assertEqual(self.events.read_text().splitlines(), ['backup'])
if __name__ == "__main__":
unittest.main()