1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
# sb-builder - ssh and virsh transport
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""The only code that talks to the VM: ssh to sbb-remote, and local virsh."""
import json
import os
import shlex
from PyQt6.QtCore import QObject, QProcess, QStandardPaths
DOWN = "ssh: connection failed"
MISSING = "sbb-remote is not installed on the VM"
def parse(rc, out, err):
"""Map one sbb-remote call to (data, error)."""
if rc == 255:
return None, DOWN
if rc == 127:
return None, MISSING
try:
data = json.loads(out)
except ValueError:
return None, err.decode(errors="replace").strip() or f"exit {rc}"
if rc != 0 or "error" in data:
return None, data.get("error", f"exit {rc}")
return data, None
class Remote(QObject):
def __init__(self, cfg, parent=None):
super().__init__(parent)
self.cfg = cfg # shared dict, updated in place on config reload
run = QStandardPaths.writableLocation(QStandardPaths.StandardLocation.RuntimeLocation)
self.control = os.path.join(run or "/tmp", "sbb-%C")
def ssh_args(self, *args):
r = self.cfg["remote"]
# ssh joins its args into a remote shell command: quote every one
return ["-o", "ControlMaster=auto", "-o", f"ControlPath={self.control}",
"-o", "ControlPersist=10m", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5",
*r["ssh_options"], r["target"], "sbb-remote", *map(shlex.quote, args)]
def call(self, cb, *args, stdin=None):
"""Run sbb-remote <args>, then cb(data, error). stdin: bytes for the helper."""
self._run("ssh", self.ssh_args(*args), lambda rc, out, err: cb(*parse(rc, out, err)),
stdin)
def virsh(self, cb, *args):
"""Run local virsh against the configured libvirt URI, then cb(rc, text)."""
argv = ["-c", self.cfg["vm"]["libvirt_uri"], *args]
self._run("virsh", argv,
lambda rc, out, err: cb(rc, (out or err).decode(errors="replace").strip()))
def follow(self, jid, offset):
"""Start job-follow; the caller wires readyReadStandardOutput and finished."""
p = QProcess(self)
p.start("ssh", self.ssh_args("job-follow", jid, str(offset)))
return p
def _run(self, prog, argv, cb, stdin=None):
p = QProcess(self)
def done(rc, _status=None):
cb(rc, bytes(p.readAllStandardOutput()), bytes(p.readAllStandardError()))
p.deleteLater()
p.finished.connect(done)
p.errorOccurred.connect(
lambda e: e == QProcess.ProcessError.FailedToStart and done(-1))
p.start(prog, argv)
if stdin is not None:
p.write(stdin)
p.closeWriteChannel()
|
