Merge pull request #527 from Vaso73/fix/lxc-update-session-resilience

Keep LXC updates running when the browser session expires
This commit is contained in:
MacRimi
2026-10-26 16:56:12 +01:00
committed by GitHub
14 changed files with 1487 additions and 43 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>
}
+132 -15
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()
@@ -81,6 +84,7 @@ export function ScriptTerminalModal({
// after dynamic imports resolve — captures the latest value without
// re-binding the closure.
const isOpenRef = useRef<boolean>(false)
const openGenerationRef = useRef(0)
const fitAddonRef = useRef<any>(null)
const sessionIdRef = useRef<string>(Math.random().toString(36).substring(2, 8))
@@ -93,6 +97,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 +135,7 @@ export function ScriptTerminalModal({
}, [onComplete])
const attemptReconnect = useCallback(() => {
if (!isOpen || isComplete || reconnectAttemptsRef.current >= 3) {
if (!isOpenRef.current || isComplete || reconnectAttemptsRef.current >= 3) {
return
}
@@ -154,7 +159,10 @@ 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 generation = openGenerationRef.current
const ticketedUrl = await getTicketedWsUrl(wsUrl)
if (!isOpenRef.current || !termRef.current || generation !== openGenerationRef.current) return
const ws = new WebSocket(ticketedUrl)
wsRef.current = ws
ws.onopen = () => {
@@ -170,9 +178,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 +205,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 +221,31 @@ const initMessage = {
try {
const msg = JSON.parse(event.data)
if (msg.type === "job_output") {
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current)
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") {
setCurrentInteraction(null)
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current)
termRef.current?.writeln(`\x1b[31m${msg.message}\x1b[0m`)
completionReceivedRef.current = true
setIsComplete(true)
ws.close()
return
}
if (backgroundUpdate && msg.type === "web_interaction" && msg.interaction === null) {
setCurrentInteraction(null)
setInteractionInput("")
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current)
return
}
if (msg.type === "web_interaction" && msg.interaction) {
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) {
@@ -231,6 +266,11 @@ const initMessage = {
return
}
if (msg.type === "script_complete") {
if (backgroundUpdate) {
setCurrentInteraction(null)
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current)
}
const exitCode = Number(msg.exit_code ?? 1)
completionReceivedRef.current = true
setIsComplete(true)
@@ -255,6 +295,7 @@ const initMessage = {
}
ws.onclose = (event) => {
if (!isOpenRef.current) return
setConnectionStatus("offline")
if (keepAliveIntervalRef.current) {
clearInterval(keepAliveIntervalRef.current)
@@ -266,13 +307,17 @@ const initMessage = {
if (!isComplete && reconnectAttemptsRef.current < 3) {
reconnectTimeoutRef.current = setTimeout(attemptReconnect, 2000)
} else {
setIsComplete(true)
onCompleteRef.current?.(-1)
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
@@ -300,12 +345,13 @@ const initMessage = {
// otherwise we attach a Terminal to a stale ref and open a WS that
// nobody reads.
const wasOpenAtCall = isOpenRef.current
const generation = ++openGenerationRef.current
const [TerminalClass, FitAddonClass] = await Promise.all([
import("xterm").then((mod) => mod.Terminal),
import("xterm-addon-fit").then((mod) => mod.FitAddon),
import("xterm/css/xterm.css"),
])
if (!wasOpenAtCall || !isOpenRef.current) return
if (!wasOpenAtCall || !isOpenRef.current || generation !== openGenerationRef.current) return
const fontSize = window.innerWidth < 768 ? 12 : 16
@@ -366,7 +412,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 +432,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 +462,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,7 +478,32 @@ const initMessage = {
try {
const msg = JSON.parse(event.data)
if (msg.type === "job_output") {
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current)
if (msg.truncated) term.writeln(t("vmLxc.updates.backgroundOutputTrimmed"))
term.write(msg.text)
outputOffsetRef.current = msg.offset
return
}
if (msg.type === "job_unavailable") {
setCurrentInteraction(null)
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current)
term.writeln(`\x1b[31m${msg.message}\x1b[0m`)
completionReceivedRef.current = true
setIsComplete(true)
ws.close()
return
}
if (backgroundUpdate && msg.type === "web_interaction" && msg.interaction === null) {
setCurrentInteraction(null)
setInteractionInput("")
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current)
return
}
if (msg.type === "web_interaction" && msg.interaction) {
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) {
@@ -450,6 +525,11 @@ const initMessage = {
return
}
if (msg.type === "script_complete") {
if (backgroundUpdate) {
setCurrentInteraction(null)
setIsWaitingNextInteraction(false)
if (waitingTimeoutRef.current) clearTimeout(waitingTimeoutRef.current)
}
const exitCode = Number(msg.exit_code ?? 1)
completionReceivedRef.current = true
setIsComplete(true)
@@ -478,6 +558,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,8 +570,13 @@ const initMessage = {
}
if (!completionReceivedRef.current && !isComplete) {
setIsComplete(true)
onCompleteRef.current?.(-1)
if (backgroundUpdate) {
term.writeln(t("vmLxc.updates.backgroundDetached"))
attemptReconnect()
} else {
setIsComplete(true)
onCompleteRef.current?.(-1)
}
}
}
@@ -552,6 +638,7 @@ const initMessage = {
if (isOpen) {
initializeTerminal()
} else {
openGenerationRef.current++
if (checkConnectionInterval.current) {
clearInterval(checkConnectionInterval.current)
}
@@ -578,6 +665,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)
@@ -586,6 +674,28 @@ const initMessage = {
}
}, [isOpen])
useEffect(() => () => {
if (!backgroundUpdate) return // Transport mode is fixed for this instance.
// Conditional parents can unmount an open modal. Detach that viewer too;
// only the background job owns its process and completion result.
isOpenRef.current = false
openGenerationRef.current++
for (const ref of [checkConnectionInterval, keepAliveIntervalRef]) {
if (ref.current) clearInterval(ref.current)
}
for (const ref of [waitingTimeoutRef, reconnectTimeoutRef]) {
if (ref.current) clearTimeout(ref.current)
}
if (wsRef.current) {
wsRef.current.onclose = null
wsRef.current.onmessage = null
wsRef.current.close()
wsRef.current = null
}
termRef.current?.dispose()
termRef.current = null
}, [])
useEffect(() => {
const updateDeviceType = () => {
const width = window.innerWidth
@@ -641,8 +751,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 +763,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 +777,10 @@ const initMessage = {
if (value === "cancel" || value === "") {
setCurrentInteraction(null)
setInteractionInput("")
if (backgroundUpdate) {
wsRef.current.send(JSON.stringify({ type: "cancel" }))
return
}
handleCloseModal()
return
}
+48 -26
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,27 +2663,37 @@ 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 {
await fetchApi(`/api/lxc-updates/${applyVmid}/applied`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
success: exitCode === 0,
run_id: applyRunId,
target: applyTarget,
requested_targets: applyTargetIds,
target_labels: applyTargetLabels,
duration_seconds: duration,
ct_name: selectedVM?.name || `CT-${applyVmid}`,
refresh_docker_inventory: applyDockerRefresh,
}),
})
if (notifyBackend) {
await fetchApi(`/api/lxc-updates/${applyVmid}/applied`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
success: exitCode === 0,
run_id: applyRunId,
target: applyTarget,
requested_targets: applyTargetIds,
target_labels: applyTargetLabels,
duration_seconds: duration,
ct_name: selectedVM?.name || `CT-${applyVmid}`,
refresh_docker_inventory: applyDockerRefresh,
}),
})
}
} catch {
// Non-fatal — the backend process hook owns finalization.
}
@@ -2697,6 +2711,12 @@ export function VirtualMachines() {
// performs the requests (the data map is only a render seed), so no
// invalidation is needed and there is no "Loading applications" flash.
void fetchLxcApps(applyVmid)
// Recovery can be opened from the job banner while another CT is selected.
// Refresh the operated guest without changing that other guest's result UI.
if (selectedVMRef.current?.vmid !== applyVmid) {
void mutate()
return
}
// Enter the "Comprobando resultado…" state on the Updates tab.
// Baseline snapshot lets a downstream useEffect detect when the
// SWR poll actually delivers the post-apply counts (as opposed
@@ -7960,6 +7980,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 +7993,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 +8009,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/"
+66
View File
@@ -134,6 +134,72 @@ 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']
# A second viewer must also dismiss a question already
# answered elsewhere, or cleared when the job ends.
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"""
+375
View File
@@ -0,0 +1,375 @@
"""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
# The shared runner restores a guest it started in its EXIT trap. Its package
# observation and `pct shutdown --timeout 60` need time after SIGTERM, too.
STOP_GRACE = 300
RESULT_TTL = 6 * 3600
MAX_OUTPUT = 256 * 1024
MAX_INPUT = 64 * 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 = ""
self._input = bytearray()
def launch(self):
master, slave = pty.openpty()
self.master = master
try:
# A viewer's paste must never block the worker's output, timeout
# or cancellation when the updater does not read its stdin.
os.set_blocking(master, False)
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:
if not self.stop_reason:
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()
self._input.clear()
self.interaction = None
self.revision += 1
try:
os.killpg(self.process.pid, signal.SIGTERM)
except ProcessLookupError:
pass
def cancel(self):
self._signal("cancelled")
def _write_input(self):
# The descriptor is nonblocking, so even a full PTY cannot hold this
# lock. Retain unwritten bytes in order until the child can read them.
with self.lock:
if not self._input or self.stop_reason:
return
try:
written = os.write(self.master, self._input[:16384])
except BlockingIOError:
return
except OSError as exc:
if exc.errno != errno.EIO:
raise
# The child can close its terminal and keep working. Losing
# stdin must not turn that update into a viewer failure.
self._input.clear()
return
del self._input[:written]
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 >= STOP_GRACE:
try:
os.killpg(self.process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
self._interactions(interactions)
with self.lock:
pending_input = bool(self._input) and running and not self.stop_reason
readable, writable, _ = select.select(
[] if eof else [self.master],
[self.master] if pending_input else [], [], 0.05)
if writable:
self._write_input()
if readable:
try:
data = os.read(self.master, 16384)
except BlockingIOError:
continue
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=STOP_GRACE)
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._input.clear()
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:
data = message.encode()
if len(self._input) + len(data) > MAX_INPUT:
raise ValueError("Terminal input buffer is full")
if not self.stop_reason:
self._input.extend(data)
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)]
+2 -1
View File
@@ -138,7 +138,7 @@ const context = {
invalidateLxcApps: () => {}, setSelectedVM: value => { context.selectionChanged = value },
alert: value => { context.alerted = value },
}
for (const name of ["Vmid", "Target", "RunId", "TargetIds", "TargetLabels", "UpdateCommand", "AppName",
for (const name of ["Vmid", "Target", "RunId", "Resume", "TargetIds", "TargetLabels", "UpdateCommand", "AppName",
"RunHelper", "AllowHelperWithCustom", "DockerStandaloneTargets", "DockerEngine", "DockerRefresh",
"BackupStorage", "StartedAt", "Open", "CheckAndUpdate", "CheckedOs"])
context[`setApply${name}`] = value => { context[`apply${name}`] = value }
@@ -205,6 +205,7 @@ for (const [target, options] of [
context.applyOpen = false
ui.openApply(100, target, options)
assert.equal(context.applyOpen, true, `explicit ${target} update opens the terminal directly`)
assert.equal(context.applyResume, false, 'an explicitly started update is distinguished from attach-only recovery')
}
assert.equal(confirmationCalls, 0, 'ordinary update entry points never open a browser confirmation')
assert(!source.includes('checkAndUpdate'), 'the abandoned combined check/install action has no UI path')
+129
View File
@@ -0,0 +1,129 @@
// 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: []}
let unmount
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))
if (ts.isCallExpression(node) && node.expression.getText(tree) === 'useEffect'
&& node.arguments[0]?.getText(tree).includes('Transport mode is fixed'))
unmount = node.arguments[0].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: value => context.waiting = value,
setCurrentInteraction: value => context.interaction = value, setInteractionInput() {},
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')
f.context.waiting = true
message({data: JSON.stringify({type: 'web_interaction', interaction: null})})
assert.equal(f.context.interaction, null, 'answer in another viewer dismisses the stale question')
assert.equal(f.context.waiting, false)
f.context.waiting = true
message({data: JSON.stringify({type: 'job_output', text: 'working', offset: 34})})
assert.equal(f.context.waiting, false, 'ordinary progress clears the post-answer spinner')
f.context.interaction = {id: 'stale'}; f.context.waiting = true
message({data: JSON.stringify({type: 'script_complete', exit_code: 3})})
assert.deepEqual(f.finished, [3])
assert.equal(f.context.interaction, null, 'a completed job cannot retain an active question')
assert.equal(f.context.waiting, false, 'a completed job cannot retain a processing overlay')
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')
}
assert.ok(unmount)
for (const backgroundUpdate of [undefined, {runId:'existing'}]) {
const f=fixture(backgroundUpdate), closes=[], sends=[], disposals=[]
const ref={current:{close:()=>closes.push(true),send:()=>sends.push(true)}}
Object.assign(f.context,{wsRef:ref,openGenerationRef:{current:1},
checkConnectionInterval:{current:null},reconnectTimeoutRef:{current:null}})
f.context.termRef.current.dispose=()=>disposals.push(true)
compile(unmount,f.context)()()
assert.equal(closes.length,backgroundUpdate?1:0)
assert.equal(sends.length,0,'unmount cannot send cancellation or another start')
assert.equal(f.finished.length,0)
assert.equal(disposals.length,backgroundUpdate?1:0)
}
const vms=fs.readFileSync(path.join(root,'components/virtual-machines.tsx'),'utf8')
const vmTree=ts.createSourceFile('vms.tsx',vms,ts.ScriptTarget.Latest,true,ts.ScriptKind.TSX)
let complete
function visit(n){if(ts.isVariableDeclaration(n)&&n.name.getText(vmTree)==='handleApplyComplete')complete=n.initializer.getText(vmTree);ts.forEachChild(n,visit)}
visit(vmTree)
async function checkCompletion(){
for(const selected of [100,200]){
const posts=[],states=[],refresh=[]
const context={applyVmid:100,applyStartedAt:Date.now(),selectedVM:{vmid:selected},selectedVMRef:{current:{vmid:selected}},
applyRunId:'existing',applyTarget:'app',applyTargetIds:['app:test'],applyTargetLabels:['Test'],applyDockerRefresh:false,
vmModalCacheRef:{current:{details:new Map(),schedule:new Map()}},fetchApi:async(...args)=>posts.push(args),
fetchLxcApps:id=>refresh.push(id),mutate:()=>{},setUpdatesBaselineCount:v=>states.push(v),setUpdatesResult:v=>states.push(v),setUpdatesRefreshing:v=>states.push(v)}
await compile(complete,context)(0,false)
assert.equal(posts.length,0,'background completion never sends a second finalization POST')
assert.deepEqual(refresh,[100])
assert.equal(states.length,selected===100?3:0,'another guest result UI is preserved')
}
}
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')
checkCompletion().catch(e=>{console.error(e);process.exitCode=1})
+640
View File
@@ -0,0 +1,640 @@
"""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_cancel_waits_for_the_runners_original_state_cleanup(self):
# apply_updates.sh's EXIT trap may need a 60s package probe followed
# by `pct shutdown --timeout 60`. A two-second kill truncates that trap.
self.script.write_text(
f"trap 'sleep 2.3; echo restored >> {shlex.quote(str(self.events))}; exit 0' TERM\n"
+ self.script.read_text())
job = self.start()
wait_for(lambda: self.events.exists())
job.cancel()
wait_for(lambda: job.state == 'complete')
self.assertEqual(job.exit_code, 130)
self.assertEqual(self.events.read_text().splitlines(), ['backup', 'restored'])
self.assertNotIn('update', self.events.read_text())
self.callback.assert_called_once()
self.assertGreater(jobs.STOP_GRACE, 120)
def test_timeout_still_reaps_a_process_that_ignores_termination(self):
self.script.write_text("trap '' TERM\necho ready\nwhile :; do sleep .02; done\n")
with patch.object(jobs, 'MAX_RUNTIME', .15), patch.object(jobs, 'STOP_GRACE', .1):
job = self.start()
wait_for(lambda: job.state == 'complete')
self.assertEqual(job.exit_code, 124)
self.assertIsNotNone(job.process.poll())
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_large_paste_cannot_block_output_cancel_or_timeout(self):
self.script.write_text("stty raw -echo\necho ready\nwhile :; do echo progress; sleep .02; done\n")
for run, cancel in [("cancel-paste", True), ("timeout-paste", False)]:
with self.subTest(run=run), patch.object(jobs, "MAX_RUNTIME", .5), \
patch.object(jobs, "STOP_GRACE", .1):
job = self.start(run)
wait_for(lambda: "ready" in job.read(0)["text"])
before = job.read(0)["offset"]
job.input("x" * 16384)
job.input("y" * 16384)
wait_for(lambda: job.read(0)["offset"] > before)
if cancel:
job.cancel()
wait_for(lambda: job.state == "complete")
self.assertEqual(job.exit_code, 130 if cancel else 124)
self.assertIsNotNone(job.process.poll())
self.assertFalse(job._input)
self.assertEqual(self.callback.call_count, 2)
def test_queued_paste_reaches_a_slow_reader_in_order_without_losing_bytes(self):
payload = ("Ž🙂abc" * 3500).encode()
received = self.folder / "received"
reader = (
"import os,pathlib; data=bytearray(); "
f"size={len(payload)}; "
"exec('while len(data)<size:\\n data.extend(os.read(0,min(257,size-len(data))))'); "
f"pathlib.Path({str(received)!r}).write_bytes(data)"
)
self.script.write_text(
"stty raw -echo\necho ready\n"
f"while [ ! -f {shlex.quote(str(self.gate))} ]; do sleep .02; done\n"
+ shlex.quote(sys.executable) + " -c " + shlex.quote(reader) + "\n")
job = self.start()
wait_for(lambda: "ready" in job.read(0)["text"])
text = payload.decode()
for offset in range(0, len(text), 8000):
job.input(text[offset:offset + 8000])
self.gate.touch()
wait_for(lambda: job.state == "complete")
self.assertEqual(job.exit_code, 0)
self.assertEqual(received.read_bytes(), payload)
self.callback.assert_called_once()
def test_full_input_queue_rejects_before_any_write_and_preserves_pending_data(self):
job = self.start()
with job.lock, patch.object(jobs.os, "write") as write:
job._input = bytearray(b"x" * jobs.MAX_INPUT)
before = bytes(job._input)
with self.assertRaisesRegex(ValueError, "buffer is full"):
job.input("overflow")
self.assertEqual(bytes(job._input), before)
write.assert_not_called()
job._input.clear()
self.assertIsNone(job.stop_reason)
self.finish(job)
def test_multibyte_output_crosses_pty_chunks_without_corrupting_replay(self):
payload = "Ž🙂abc" * 5000
command = "import os; os.write(1," + repr(payload) + ".encode())"
self.script.write_text(shlex.quote(sys.executable) + " -c " + shlex.quote(command) + "\n")
job = self.start()
wait_for(lambda: job.state == "complete")
self.assertEqual(job.read(0)["text"], payload)
self.assertEqual(job.read(len(payload) - 4)["text"], payload[-4:])
def test_capacity_rejection_preserves_running_jobs_and_evicts_only_completed(self):
with patch.object(jobs, "MAX_JOBS", 2):
first, second = self.start(), self.start("second", "101")
before = set(jobs._jobs)
with patch.object(jobs.subprocess, "Popen") as spawn:
with self.assertRaises(jobs.JobConflict):
self.start("third", "102")
spawn.assert_not_called()
self.assertEqual(set(jobs._jobs), before)
with second.lock:
self.finish(first)
wait_for(lambda: second.state == "complete")
third = self.start("third", "102")
wait_for(lambda: third.state == "complete")
self.assertNotIn(first.id, jobs._jobs)
self.assertIn(second.id, jobs._jobs)
self.assertIn(third.id, jobs._jobs)
def test_finalization_keeps_the_guest_reserved_and_callback_runs_once(self):
entered, release = threading.Event(), threading.Event()
self.addCleanup(release.set)
def finalize(*_):
entered.set()
release.wait(3)
self.callback.side_effect = finalize
job = self.start()
self.gate.touch()
self.assertTrue(entered.wait(3))
self.assertEqual(job.state, "finalizing")
with patch.object(jobs.subprocess, "Popen") as spawn:
with self.assertRaises(jobs.JobConflict):
self.start("other")
self.assertIs(jobs.get(job.id, 100), job)
spawn.assert_not_called()
release.set()
wait_for(lambda: job.state == "complete")
self.callback.assert_called_once()
def test_worker_launch_failure_reaps_the_process_without_registering_a_job(self):
children = []
real_spawn = subprocess.Popen
def spawn(*args, **kwargs):
child = real_spawn(*args, **kwargs)
children.append(child)
return child
with patch.object(jobs.subprocess, "Popen", side_effect=spawn), \
patch.object(jobs.threading.Thread, "start", side_effect=RuntimeError("no worker")):
with self.assertRaisesRegex(RuntimeError, "no worker"):
self.start()
self.assertFalse(jobs.list_jobs())
self.assertEqual(len(children), 1)
self.assertIsNotNone(children[0].poll())
self.callback.assert_not_called()
def test_interaction_file_collision_is_not_overwritten_or_accepted(self):
job = self.start()
identity = "fixture-" + os.urandom(8).hex()
response = Path("/tmp/proxmenux_response_" + identity)
self.addCleanup(lambda: response.unlink(missing_ok=True))
response.symlink_to(self.events)
job.interaction = dict(type="input", id=identity)
wait_for(lambda: self.events.exists())
before = self.events.read_bytes() if self.events.exists() else None
with self.assertRaises(OSError):
job.input(dict(type="interaction_response", id=identity, value="do-not-write"))
self.assertEqual(job.interaction["id"], identity)
self.assertEqual(self.events.read_bytes() if self.events.exists() else None, before)
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_answer_in_one_viewer_clears_the_question_for_all_viewers(self):
self.attach(dict(resume=False, script_path=str(self.script), params=self.params()),
self.ticket().json["ticket"])
job = jobs.get("run-1", 100)
identity = "fixture-" + os.urandom(8).hex()
response = Path("/tmp/proxmenux_response_" + identity)
self.addCleanup(lambda: response.unlink(missing_ok=True))
job.interaction = dict(type="yesno", id=identity, title="Question", message="Continue?")
job.revision += 1
first = self.attach(dict(resume=True, vmid=100), self.ticket().json["ticket"],
[json.dumps(dict(type="interaction_response", id=identity, value="yes"))])
second = self.attach(dict(resume=True, vmid=100), self.ticket().json["ticket"])
for ws in [first, second]:
frames = [json.loads(c.args[0]) for c in ws.send.call_args_list]
self.assertIn(dict(type="web_interaction", interaction=None), frames)
self.assertEqual(response.read_text(), "yes")
self.assertIsNone(job.stop_reason)
self.callback.assert_not_called()
self.finish(job)
self.callback.assert_called_once()
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()
+1 -1
View File
@@ -23,7 +23,7 @@ function executable(node, file, scope) {
function fixture(handler, locale, optedIn=true) {
const writes=[], callbacks=[], state={closed:0};
const t=translator(locale), terminal={write:s=>writes.push(s),writeln:s=>writes.push(s)};
const shared={...completion, cancelledExitCode:optedIn?6:undefined,t,
const shared={...completion, backgroundUpdate:undefined, isOpenRef:{current:true}, openGenerationRef:{current:1}, cancelledExitCode:optedIn?6:undefined,t,
completedSuccessfullyMessage:undefined,completedWithErrorMessage:undefined};
const getCompletionMessage=executable(messageCallback,source,shared);
const scope={...shared,getCompletionMessage,term:terminal,termRef:{current:terminal},