Snapshot-Explorer: Dateien zurueckholen, im Terminal und im Browser

Snapshots anzulegen half bisher nur halb - an die Daten darin kam man nur
ueber einen Rollback. Dafuer jetzt zwei Werkzeuge auf gemeinsamer Basis.

snapfs.py bindet einen Snapshot schreibgeschuetzt ein und liefert einen
gewoehnlichen Pfad; die Oberflaechen wissen dadurch nichts ueber Storages:

  rbd (Ceph)   rbd map pool/image@snap, bei nicht unterstuetzten
               Image-Features faellt es auf rbd-nbd zurueck
  zfspool      Container ueber .zfs/snapshot, VMs ueber einen Klon
  lvmthin/lvm  die Snapshot-LV snap_<volume>_<snapname> aktivieren
  dir/nfs/cifs qemu-nbd --load-snapshot (nur qcow2)

Gemountet wird mit ro,noload bzw. ro,norecovery,nouuid - Snapshots
laufender Gaeste haben fast immer ein unsauberes Journal. Alles Angelegte
steht in /run/pvesnap/explorer.json und laesst sich nach einem Absturz mit
"--cleanup" wieder abraeumen.

pvesnap-explorer: zwei Fenster wie im Midnight Commander, links der
Snapshot, rechts der lokale Rechner. Markieren, F5, Fortschrittsbalken,
ESC bricht ab, vorhandene Dateien werden abgefragt. In den Snapshot hinein
kann nicht kopiert werden. Geraetedateien und Sockets werden
uebersprungen, symbolische Verweise bleiben Verweise.

pvesnap-web: derselbe Inhalt im Browser, auch vom anderen Rechner.
Einzelne Dateien direkt, Verzeichnisse und Mehrfachauswahl als ZIP, das im
Strom erzeugt wird - ohne Zwischendatei auf der Platte. Zugang nur mit dem
beim Start ausgegebenen Schluessel; Pfade ausserhalb des Snapshots werden
abgewiesen; es wird ausschliesslich gelesen.

Nebenbei: die curses-Bausteine sind aus tui.py nach curses_util.py
gewandert, damit Editor und Explorer sie teilen.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
duffyduck
2026-07-31 09:14:16 +02:00
co-authored by Claude Opus 5
parent 1be13ad294
commit f37a942fea
11 changed files with 2839 additions and 223 deletions
+110 -2
View File
@@ -12,6 +12,11 @@ löscht. Alles wird über **eine INI-Datei** gesteuert, die sich auch bequem in
* Läuft cluster-weit (alle Nodes) über `pvesh`
* Nur Python-Standardbibliothek — keine zusätzlichen Pakete nötig
Dazu zwei Werkzeuge, um wieder **an die Daten im Snapshot** zu kommen:
* `pvesnap-explorer` — zwei Fenster im Terminal wie im Midnight Commander
* `pvesnap-web` — im Browser durchsehen und herunterladen, auch vom anderen Rechner
---
## Installation
@@ -28,7 +33,9 @@ Das Skript legt an:
| Pfad | Inhalt |
|---|---|
| `/usr/lib/pvesnap/` | Programmcode |
| `/usr/local/bin/pvesnap` | Startbefehl |
| `/usr/local/bin/pvesnap` | Dienst und Konfiguration |
| `/usr/local/bin/pvesnap-explorer` | Zwei-Fenster-Explorer |
| `/usr/local/bin/pvesnap-web` | Web-Oberfläche |
| `/etc/pvesnap/pvesnap.conf` | Konfiguration (wird bei Updates **nicht** überschrieben) |
| `/var/lib/pvesnap/state.json` | merkt sich die letzten Läufe |
| `/etc/systemd/system/pvesnap.service` | systemd-Unit |
@@ -320,6 +327,96 @@ oder ein abgebrochener Task, der die Sperre nicht freigegeben hat.
---
## Dateien aus einem Snapshot holen
Ein Snapshot ist kein Verzeichnis. Damit man hineinschauen kann, wird er
**schreibgeschützt** als Blockgerät eingebunden und gemountet — je nach Storage
unterschiedlich:
| Storage | Weg hinein |
|---|---|
| `rbd` (Ceph) | `rbd map pool/image@snap`, bei nicht unterstützten Image-Features `rbd-nbd` |
| `zfspool` | Container direkt über `.zfs/snapshot/…`, VMs über einen temporären Klon |
| `lvmthin`, `lvm` | die Snapshot-LV `snap_<volume>_<snapname>` aktivieren |
| `dir`, `nfs`, `cifs` | `qemu-nbd --load-snapshot` (nur qcow2) |
Danach werden Partitionen erkannt und mit `ro,noload` bzw. `ro,norecovery,nouuid`
gemountet — Snapshots laufender Gäste haben fast immer ein unsauberes Journal,
ohne diese Optionen würde der Kernel hineinschreiben wollen.
Aufgeräumt wird beim Beenden automatisch. Bleibt nach einem Absturz etwas
hängen:
```bash
pvesnap-explorer --cleanup
```
### `pvesnap-explorer` — zwei Fenster im Terminal
```bash
pvesnap-explorer # Gast und Snapshot auswählen
pvesnap-explorer 802 # Snapshot von VM 802 auswählen
pvesnap-explorer 802 <snapname> # direkt öffnen
pvesnap-explorer --path /mnt/x # beliebiges Verzeichnis (zum Ausprobieren)
```
Links der Snapshot, rechts der lokale Rechner:
| Taste | |
|---|---|
| `Tab` | Fenster wechseln |
| `Pfeiltasten`, `Enter` | bewegen, Verzeichnis öffnen oder Datei ansehen |
| `Backspace` | ein Verzeichnis hoch |
| `Leertaste` | markieren; `*` umkehren, `a` alle, `u` keine |
| `F5` / `c` | Markiertes ins andere Fenster kopieren |
| `F3` / `v` | Datei ansehen (Text oder Hex) |
| `F7` / `n` | neues Verzeichnis (nur lokal) |
| `F6` / `g` | Verzeichnis direkt eingeben |
| `F2` / `m` | anderes Dateisystem des Snapshots wählen |
| `F10` / `q` | beenden |
Kopiert wird immer vom aktiven ins andere Fenster — in den Snapshot hinein geht
nichts, er ist schreibgeschützt. Beim Kopieren gibt es einen Fortschrittsbalken,
`ESC` bricht ab; vorhandene Dateien werden abgefragt (`ja` / `nein` / `alle` /
`keine`). Gerätedateien und Sockets werden übersprungen, symbolische Verweise
bleiben Verweise.
### `pvesnap-web` — im Browser herunterladen
```bash
pvesnap-web # Gast und Snapshot im Browser wählen
pvesnap-web 802 <snapname> # diesen Snapshot gleich öffnen
pvesnap-web --port 8080 --bind 10.0.0.5
```
Beim Start wird eine Adresse mit Zugangsschlüssel ausgegeben:
```
http://192.168.1.50:8823/?token=zGo-1RW7eYyf_YCPvTIOLtOp
```
Diese Adresse an einem anderen Rechner im Browser öffnen. Dort lässt sich der
Snapshot durchklicken; **einzelne Dateien** werden direkt heruntergeladen,
**Verzeichnisse und Mehrfachauswahl** als ZIP (im Strom erzeugt, es wird nichts
zwischengespeichert).
Zur Sicherheit:
* Ohne den Zugangsschlüssel gibt es nichts zu sehen (er wandert nach dem ersten
Aufruf in ein Cookie). `--no-token` schaltet das ab — nur in einem Netz
benutzen, dem du vertraust.
* Es wird ausschließlich **gelesen**. Die Oberfläche kann nichts ändern oder
löschen.
* Pfade außerhalb des eingebundenen Snapshots werden abgewiesen — über die
Adresszeile kommt man nicht an den Rest des Hosts.
* Die Verbindung ist unverschlüsseltes HTTP. Über unsichere Netze besser einen
SSH-Tunnel legen:
`ssh -L 8823:localhost:8823 root@pve` und dann `pvesnap-web --bind 127.0.0.1`.
Beendet wird mit `Strg-C`; der Snapshot wird dabei wieder ausgehängt.
---
## Aufbau
```
@@ -331,9 +428,20 @@ pvesnap/
engine.py Auswahl der Gäste, Anlegen, Aufräumen
daemon.py Hauptschleife, Signale, Sperren
state.py merkt sich die letzten Läufe
tui.py ncurses-Editor
preflight.py prüft die Umgebung (root, /etc/pve beschreibbar, /root)
cli.py Kommandozeile
curses_util.py gemeinsame curses-Bausteine
tui.py ncurses-Konfigurationseditor
snapfs.py Snapshots einbinden und mounten (rbd/zfs/lvm/qcow2)
explorer.py Zwei-Fenster-Explorer im Terminal
web/ Web-Oberfläche zum Herunterladen
server.py HTTP-Server, ZIP im Strom, Zugangsschlüssel
assets.py Stylesheet und HTML-Gerüst
config/pvesnap.conf.example
systemd/pvesnap.service
install.sh uninstall.sh
```
Der Dienst (`pvesnap`) und die beiden Explorer teilen sich `proxmox.py`;
`snapfs.py` liefert beiden Oberflächen einen ganz gewöhnlichen Pfad, sodass
Explorer und Web-Oberfläche nichts über Ceph, ZFS oder LVM wissen müssen.
+14 -4
View File
@@ -61,13 +61,19 @@ cp -r "${SRC}/pvesnap" "${LIB_DIR}/pvesnap"
find "${LIB_DIR}/pvesnap" -name '__pycache__' -type d -exec rm -rf {} + 2>/dev/null || true
chmod -R go-w "${LIB_DIR}/pvesnap"
info "Installiere Startbefehl ${BIN}"
cat > "${BIN}" <<EOF
info "Installiere Startbefehle"
install_command() { # $1 = Befehlsname, $2 = Python-Modul
cat > "/usr/local/bin/$1" <<EOF
#!/bin/sh
# von install.sh erzeugt
PYTHONPATH="${LIB_DIR}\${PYTHONPATH:+:\$PYTHONPATH}" exec /usr/bin/python3 -m pvesnap "\$@"
PYTHONPATH="${LIB_DIR}\${PYTHONPATH:+:\$PYTHONPATH}" exec /usr/bin/python3 -m $2 "\$@"
EOF
chmod 0755 "${BIN}"
chmod 0755 "/usr/local/bin/$1"
info " /usr/local/bin/$1"
}
install_command pvesnap pvesnap
install_command pvesnap-explorer pvesnap.explorer
install_command pvesnap-web pvesnap.web
# --- Konfiguration ----------------------------------------------------
install -d -m 0755 "${CONF_DIR}" "${STATE_DIR}" "${DOC_DIR}"
@@ -122,6 +128,10 @@ Fertig.
Testlauf ohne Aenderung : pvesnap run --force --dry-run
Nach Aenderungen : systemctl reload pvesnap
Protokoll : journalctl -u pvesnap -f
Dateien aus einem Snapshot holen:
pvesnap-explorer zwei Fenster wie im Midnight Commander
pvesnap-web im Browser durchsehen und herunterladen
EOF
if [ "${NEW_CONFIG:-0}" -eq 1 ]; then
+227
View File
@@ -0,0 +1,227 @@
"""Gemeinsame curses-Bausteine fuer Konfigurationseditor und Explorer."""
from __future__ import annotations
import curses
C_HEADER, C_FOOTER, C_SEL, C_WARN, C_OK, C_DIM, C_DIR, C_MARK = 1, 2, 3, 4, 5, 6, 7, 8
def init_colors():
try:
curses.start_color()
curses.use_default_colors()
curses.init_pair(C_HEADER, curses.COLOR_WHITE, curses.COLOR_BLUE)
curses.init_pair(C_FOOTER, curses.COLOR_BLACK, curses.COLOR_WHITE)
curses.init_pair(C_SEL, curses.COLOR_BLACK, curses.COLOR_CYAN)
curses.init_pair(C_WARN, curses.COLOR_WHITE, curses.COLOR_RED)
curses.init_pair(C_OK, curses.COLOR_BLACK, curses.COLOR_GREEN)
curses.init_pair(C_DIM, -1, -1)
curses.init_pair(C_DIR, curses.COLOR_CYAN, -1)
curses.init_pair(C_MARK, curses.COLOR_YELLOW, -1)
except curses.error:
pass
# ---------------------------------------------------------------------------
# Zeichnen
# ---------------------------------------------------------------------------
def put(win, y, x, text, attr=0):
"""addstr, das bei zu kleinem Terminal nicht abstuerzt."""
height, width = win.getmaxyx()
if y < 0 or y >= height or x >= width:
return
try:
win.addstr(y, x, str(text)[: max(0, width - x - 1)], attr)
except curses.error:
pass
def fill(win, y, text, attr):
height, width = win.getmaxyx()
if y < 0 or y >= height:
return
try:
win.addstr(y, 0, str(text).ljust(width - 1)[: width - 1], attr)
except curses.error:
pass
def fill_row(win, y, x, text, width, attr):
put(win, y, x, str(text).ljust(max(0, width - x - 1))[: max(0, width - x - 1)], attr)
# ---------------------------------------------------------------------------
# Tasten
# ---------------------------------------------------------------------------
def read_key(win):
try:
return win.get_wch()
except curses.error:
return None
except KeyboardInterrupt:
return "\x1b"
def is_enter(key):
return key in ("\n", "\r", curses.KEY_ENTER)
def is_escape(key):
return key == "\x1b"
def is_backspace(key):
return key in ("\x7f", "\b", "\x08", curses.KEY_BACKSPACE)
def keybar(width, items):
"""Tastenleiste als Liste von Zeilen, passend zur Breite des Terminals.
Reicht der Platz nicht fuer eine Zeile, wird auf zwei Zeilen umgebrochen;
bei ganz schmalen Terminals bleiben nur noch die Tasten uebrig.
"""
def join(entries, index):
return " | ".join("%s %s" % (entry[0], entry[index]) for entry in entries)
long_text = join(items, 1)
if len(long_text) + 2 <= width:
return [long_text]
half = (len(items) + 1) // 2
first, second = join(items[:half], 1), join(items[half:], 1)
if max(len(first), len(second)) + 2 <= width:
return [first, second]
short_text = join(items, 2)
if len(short_text) + 2 <= width:
return [short_text]
first, second = join(items[:half], 2), join(items[half:], 2)
if max(len(first), len(second)) + 2 <= width:
return [first, second]
return [" ".join(entry[0] for entry in items)]
# ---------------------------------------------------------------------------
# Dialoge
# ---------------------------------------------------------------------------
def message(win, text, error=False):
height, _ = win.getmaxyx()
attr = curses.color_pair(C_WARN if error else C_OK) | curses.A_BOLD
fill(win, height - 2, " " + str(text) + " [Taste druecken]", attr)
win.refresh()
read_key(win)
def confirm(win, question):
height, _ = win.getmaxyx()
fill(win, height - 2, " %s [j/n]" % question,
curses.color_pair(C_WARN) | curses.A_BOLD)
win.refresh()
while True:
key = read_key(win)
if key in ("j", "J", "y", "Y"):
return True
if key in ("n", "N") or is_escape(key):
return False
def ask(win, question, answers, default=None):
"""Mehrfachauswahl in einer Zeile. `answers` ist z.B. 'jnak'."""
height, _ = win.getmaxyx()
fill(win, height - 2, " %s" % question, curses.color_pair(C_WARN) | curses.A_BOLD)
win.refresh()
while True:
key = read_key(win)
if isinstance(key, str) and key.lower() in answers:
return key.lower()
if is_escape(key):
return default
def prompt(win, label, initial=""):
"""Einzeilige Eingabe am unteren Rand. None = abgebrochen."""
height, width = win.getmaxyx()
buffer = list(str(initial))
curses.curs_set(1)
try:
while True:
text = "".join(buffer)
label_text = " %s: " % label
visible = width - len(label_text) - 2
shown = text[-visible:] if visible > 0 and len(text) > visible else text
fill(win, height - 2, label_text + shown, curses.color_pair(C_SEL))
try:
win.move(height - 2, min(width - 2, len(label_text) + len(shown)))
except curses.error:
pass
win.refresh()
key = read_key(win)
if key is None:
continue
if is_enter(key):
return "".join(buffer).strip()
if is_escape(key):
return None
if is_backspace(key):
if buffer:
buffer.pop()
continue
if key == "\x15": # Strg-U
buffer = []
continue
if isinstance(key, str) and key.isprintable():
buffer.append(key)
finally:
curses.curs_set(0)
def choose(win, title, entries, current=0):
"""Kleines Auswahlfenster. entries: Liste von (wert, beschriftung)."""
if not entries:
return None
height, width = win.getmaxyx()
box_h = min(len(entries) + 4, height - 2)
box_w = min(max([len(title)] + [len(e[1]) for e in entries]) + 8, width - 2)
top = max(0, (height - box_h) // 2)
left = max(0, (width - box_w) // 2)
box = curses.newwin(box_h, box_w, top, left)
box.keypad(True)
index = max(0, min(current, len(entries) - 1))
scroll = 0
rows = max(1, box_h - 4)
while True:
if index < scroll:
scroll = index
if index >= scroll + rows:
scroll = index - rows + 1
box.erase()
box.box()
put(box, 0, 2, " %s " % title, curses.A_BOLD)
for row in range(rows):
position = scroll + row
if position >= len(entries):
break
attr = curses.color_pair(C_SEL) if position == index else 0
put(box, row + 2, 2, " " + entries[position][1].ljust(box_w - 5), attr)
box.refresh()
key = read_key(box)
if key in (curses.KEY_UP, "k"):
index = (index - 1) % len(entries)
elif key in (curses.KEY_DOWN, "j"):
index = (index + 1) % len(entries)
elif key == curses.KEY_NPAGE:
index = min(len(entries) - 1, index + rows)
elif key == curses.KEY_PPAGE:
index = max(0, index - rows)
elif is_enter(key):
return entries[index][0]
elif is_escape(key) or key == "q":
return None
+930
View File
@@ -0,0 +1,930 @@
"""pvesnap-explorer - Dateien aus einem Snapshot holen.
Zwei Fenster wie im Midnight Commander: links der geoeffnete Snapshot
(schreibgeschuetzt), rechts der lokale Rechner. Markieren, F5 - fertig.
pvesnap-explorer Gast und Snapshot auswaehlen
pvesnap-explorer 802 Snapshot von VM 802 auswaehlen
pvesnap-explorer 802 <snapname> direkt oeffnen
pvesnap-explorer --path /mnt beliebiges Verzeichnis (ohne Proxmox)
pvesnap-explorer --cleanup Reste eines Absturzes aufraeumen
"""
from __future__ import annotations
import argparse
import curses
import errno
import logging
import os
import shutil
import stat
import sys
import time
from dataclasses import dataclass, field
from datetime import datetime
from . import __version__
from .curses_util import (C_DIM, C_DIR, C_FOOTER, C_HEADER, C_MARK, C_OK, C_SEL,
C_WARN, ask, choose, confirm, fill, init_colors, is_enter,
is_escape, keybar, message, prompt, put, read_key)
from .proxmox import Proxmox, ProxmoxError, pvesh_available
from .snapfs import (Session, SnapfsError, cleanup_leftovers, human_bytes,
list_snapshots)
from .util import truncate
log = logging.getLogger("pvesnap.explorer")
KEYS = [
("Tab", "Fenster", "Fenst."),
("Enter", "Oeffnen", "Oeffn."),
("Leer", "Markieren", "Mark."),
("F5", "Kopieren", "Kopie"),
("F3", "Ansehen", "Ansehen"),
("F7", "Neuer Ordner", "Ordner"),
("F2", "Laufwerk", "Laufw."),
("F10", "Ende", "Ende"),
]
COPY_BUFFER = 1024 * 1024
# ---------------------------------------------------------------------------
# Eintraege und Fenster
# ---------------------------------------------------------------------------
@dataclass
class Entry:
name: str
path: str
is_dir: bool = False
size: int = 0
mtime: float = 0.0
is_link: bool = False
is_special: bool = False
is_up: bool = False
@property
def sort_key(self):
return (0 if self.is_up else (1 if self.is_dir else 2), self.name.lower())
@dataclass
class Pane:
path: str
root: str = "/"
readonly: bool = False
title: str = ""
entries: list = field(default_factory=list)
index: int = 0
top: int = 0
marks: set = field(default_factory=set)
error: str = ""
def reload(self):
previous = self.current.name if self.entries and 0 <= self.index < len(self.entries) else ""
self.entries = []
self.error = ""
real = os.path.realpath(self.path)
if real != os.path.realpath(self.root):
parent = os.path.dirname(real) or "/"
self.entries.append(Entry(name="..", path=parent, is_dir=True, is_up=True))
try:
with os.scandir(self.path) as scan:
for item in scan:
self.entries.append(self._entry(item))
except OSError as exc:
self.error = "%s: %s" % (self.path, exc.strerror or exc)
self.entries.sort(key=lambda e: e.sort_key)
self.marks = {name for name in self.marks
if any(e.name == name for e in self.entries)}
self.index = 0
for position, entry in enumerate(self.entries):
if entry.name == previous:
self.index = position
break
self._clamp()
@staticmethod
def _entry(item):
try:
info = item.stat(follow_symlinks=False)
mode = info.st_mode
is_link = stat.S_ISLNK(mode)
if is_link:
try:
target = item.stat(follow_symlinks=True)
is_dir = stat.S_ISDIR(target.st_mode)
except OSError:
is_dir = False
special = False
else:
is_dir = stat.S_ISDIR(mode)
special = not (is_dir or stat.S_ISREG(mode))
return Entry(name=item.name, path=item.path, is_dir=is_dir,
size=0 if is_dir else info.st_size, mtime=info.st_mtime,
is_link=is_link, is_special=special)
except OSError:
return Entry(name=item.name, path=item.path)
# -- Navigation -------------------------------------------------------
@property
def current(self):
if 0 <= self.index < len(self.entries):
return self.entries[self.index]
return Entry(name="", path=self.path)
def _clamp(self):
self.index = max(0, min(self.index, max(0, len(self.entries) - 1)))
def move(self, delta):
self.index += delta
self._clamp()
def open_current(self):
entry = self.current
if not entry.is_dir:
return False
target = os.path.realpath(entry.path)
root = os.path.realpath(self.root)
if not (target == root or target.startswith(root + os.sep)):
return False # nicht aus dem Snapshot herauslaufen
self.path = target
self.marks.clear()
self.reload()
return True
def go(self, path):
if os.path.isdir(path):
self.path = os.path.realpath(path)
self.marks.clear()
self.reload()
return True
return False
def toggle_mark(self):
entry = self.current
if entry.is_up or not entry.name:
return
if entry.name in self.marks:
self.marks.discard(entry.name)
else:
self.marks.add(entry.name)
def selection(self):
"""Markierte Eintraege, sonst der Eintrag unter dem Balken."""
if self.marks:
return [e for e in self.entries if e.name in self.marks]
entry = self.current
return [] if (entry.is_up or not entry.name) else [entry]
def free_space(self):
try:
stats = os.statvfs(self.path)
return stats.f_bavail * stats.f_frsize
except OSError:
return 0
# ---------------------------------------------------------------------------
# Kopieren
# ---------------------------------------------------------------------------
class Aborted(Exception):
"""Der Benutzer hat abgebrochen."""
class Copier:
"""Kopiert Dateien und Verzeichnisse mit Fortschritt und Abbruch."""
def __init__(self, win, progress):
self.win = win
self.progress = progress # Rueckruf(text, done, total)
self.total = 0
self.done = 0
self.overwrite = None # None | "alle" | "keine"
self.copied = 0
self.skipped = []
self.failed = []
# -- Groesse vorab ----------------------------------------------------
def measure(self, entries):
total = 0
for entry in entries:
if entry.is_dir and not entry.is_link:
for base, _dirs, files in os.walk(entry.path, onerror=None):
for name in files:
try:
info = os.lstat(os.path.join(base, name))
if stat.S_ISREG(info.st_mode):
total += info.st_size
except OSError:
pass
elif not entry.is_special:
total += entry.size
self.total = total
return total
# -- Ausfuehren -------------------------------------------------------
def copy(self, entries, target):
for entry in entries:
self._copy_any(entry.path, os.path.join(target, entry.name))
def _copy_any(self, source, destination):
self._check_abort()
try:
info = os.lstat(source)
except OSError as exc:
self.failed.append("%s: %s" % (source, exc.strerror or exc))
return
if stat.S_ISLNK(info.st_mode):
self._copy_symlink(source, destination)
elif stat.S_ISDIR(info.st_mode):
self._copy_dir(source, destination)
elif stat.S_ISREG(info.st_mode):
self._copy_file(source, destination, info)
else:
# Geraetedateien, Sockets, FIFOs - im Gast-Dateisystem haeufig,
# ausserhalb aber sinnlos.
self.skipped.append("%s (Spezialdatei)" % source)
def _copy_symlink(self, source, destination):
try:
target = os.readlink(source)
if os.path.lexists(destination):
if not self._may_overwrite(destination):
return
os.remove(destination)
os.symlink(target, destination)
self.copied += 1
except OSError as exc:
self.failed.append("%s: %s" % (source, exc.strerror or exc))
def _copy_dir(self, source, destination):
try:
os.makedirs(destination, exist_ok=True)
except OSError as exc:
self.failed.append("%s: %s" % (destination, exc.strerror or exc))
return
try:
names = sorted(os.listdir(source))
except OSError as exc:
self.failed.append("%s: %s" % (source, exc.strerror or exc))
return
for name in names:
self._copy_any(os.path.join(source, name), os.path.join(destination, name))
try:
shutil.copystat(source, destination)
except OSError:
pass
def _copy_file(self, source, destination, info):
if os.path.lexists(destination) and not self._may_overwrite(destination):
self.done += info.st_size
return
self.progress(os.path.basename(source), self.done, self.total)
try:
with open(source, "rb") as src, open(destination, "wb") as dst:
while True:
self._check_abort()
chunk = src.read(COPY_BUFFER)
if not chunk:
break
dst.write(chunk)
self.done += len(chunk)
self.progress(os.path.basename(source), self.done, self.total)
shutil.copystat(source, destination)
self.copied += 1
except Aborted:
self._remove_partial(destination)
raise
except OSError as exc:
self._remove_partial(destination)
if exc.errno == errno.ENOSPC:
raise SnapfsError("Kein Platz mehr auf dem Ziel-Dateisystem.")
self.failed.append("%s: %s" % (source, exc.strerror or exc))
@staticmethod
def _remove_partial(path):
try:
os.remove(path)
except OSError:
pass
# -- Rueckfragen ------------------------------------------------------
def _may_overwrite(self, destination):
if self.overwrite == "alle":
return True
if self.overwrite == "keine":
self.skipped.append("%s (vorhanden)" % destination)
return False
answer = ask(self.win,
"%s gibt es schon: [j]a ueberschreiben, [n]ein, "
"[a]lle, [k]eine" % truncate(os.path.basename(destination), 40),
"jnak", default="n")
if answer == "a":
self.overwrite = "alle"
return True
if answer == "k":
self.overwrite = "keine"
if answer == "j":
return True
self.skipped.append("%s (vorhanden)" % destination)
return False
def _check_abort(self):
self.win.nodelay(True)
try:
key = self.win.getch()
except curses.error:
key = -1
finally:
self.win.nodelay(False)
if key == 27: # ESC
raise Aborted()
# ---------------------------------------------------------------------------
# Oberflaeche
# ---------------------------------------------------------------------------
class Explorer:
def __init__(self, left, right, session=None, subtitle=""):
self.panes = [left, right]
self.active = 0
self.session = session
self.subtitle = subtitle
self.status = ""
@property
def source(self):
return self.panes[self.active]
@property
def target(self):
return self.panes[1 - self.active]
# -- Zeichnen ---------------------------------------------------------
def draw(self, win):
win.erase()
height, width = win.getmaxyx()
keys = keybar(width, KEYS)
list_top = 3
list_bottom = height - 3 - len(keys)
rows = max(1, list_bottom - list_top)
fill(win, 0, " pvesnap-explorer - %s" % self.subtitle,
curses.color_pair(C_HEADER) | curses.A_BOLD)
middle = width // 2
for row in range(1, list_bottom + 1):
put(win, row, middle, "", curses.color_pair(C_DIM))
self._draw_pane(win, self.panes[0], 0, middle, list_top, rows, self.active == 0)
self._draw_pane(win, self.panes[1], middle + 1, width - middle - 2,
list_top, rows, self.active == 1)
entry = self.source.current
info = self.status or (entry.path if entry.name else self.source.path)
fill(win, list_bottom + 1, " " + truncate(info, width - 2),
curses.color_pair(C_DIM))
for offset, line in enumerate(reversed(keys)):
fill(win, height - 1 - offset, " " + line, curses.color_pair(C_FOOTER))
def _draw_pane(self, win, pane, x, width, top, rows, active):
if width < 8:
return
header = pane.title or pane.path
attr = curses.color_pair(C_SEL) if active else curses.color_pair(C_DIM)
put(win, 1, x, (" " + truncate(header, width - 2)).ljust(width), attr)
put(win, 2, x, truncate(pane.path, width - 1).ljust(width),
curses.A_BOLD if active else 0)
if pane.error:
put(win, top, x + 1, truncate(pane.error, width - 2),
curses.color_pair(C_WARN))
return
if pane.index < pane.top:
pane.top = pane.index
if pane.index >= pane.top + rows:
pane.top = pane.index - rows + 1
pane.top = max(0, min(pane.top, max(0, len(pane.entries) - rows)))
for row in range(rows):
position = pane.top + row
if position >= len(pane.entries):
break
entry = pane.entries[position]
marked = entry.name in pane.marks
if active and position == pane.index:
attribute = curses.color_pair(C_SEL)
elif marked:
attribute = curses.color_pair(C_MARK) | curses.A_BOLD
elif entry.is_dir:
attribute = curses.color_pair(C_DIR)
else:
attribute = 0
put(win, top + row, x, self._format(entry, width, marked), attribute)
total = len(pane.entries)
put(win, top + rows, x, " %d Eintraege%s"
% (total, " %d markiert" % len(pane.marks) if pane.marks else ""),
curses.color_pair(C_DIM))
@staticmethod
def _format(entry, width, marked):
"""Eine Zeile mit genau `width` Zeichen - laenger wuerde ins andere
Fenster hineinschreiben."""
mark = "*" if marked else " "
suffix = "@" if entry.is_link else ""
def sized(text):
return text.ljust(width)[:width]
if entry.is_up:
return sized("%s.. <hoch>" % mark)
if entry.is_dir:
size = "<DIR>"
elif entry.is_special:
size = "<SPEZ>"
else:
size = human_bytes(entry.size)
when = (datetime.fromtimestamp(entry.mtime).strftime("%d.%m.%y %H:%M")
if entry.mtime else "")
# 1 Markierung + Name + 1 + 8 Groesse + 1 + 14 Datum
room = width - 25
if room < 8:
return sized("%s%s" % (mark, truncate(entry.name + suffix, max(1, width - 2))))
name = truncate(entry.name + suffix, room).ljust(room)
return sized("%s%s %s %s" % (mark, name, size.rjust(8), when.rjust(14)))
# -- Hauptschleife ----------------------------------------------------
def run(self, win):
for pane in self.panes:
pane.reload()
while True:
self.draw(win)
win.refresh()
key = read_key(win)
rows = max(1, win.getmaxyx()[0] - 7)
self.status = ""
pane = self.source
if key == "\t" or key == curses.KEY_BTAB:
self.active = 1 - self.active
elif key in (curses.KEY_UP, "k"):
pane.move(-1)
elif key in (curses.KEY_DOWN, "j"):
pane.move(1)
elif key == curses.KEY_NPAGE:
pane.move(rows)
elif key == curses.KEY_PPAGE:
pane.move(-rows)
elif key == curses.KEY_HOME:
pane.index = 0
elif key == curses.KEY_END:
pane.index = len(pane.entries) - 1
elif is_enter(key):
if not pane.open_current():
self._view(win, pane.current)
elif key in (curses.KEY_BACKSPACE, "\x7f", curses.KEY_LEFT):
if pane.entries and pane.entries[0].is_up:
pane.index = 0
pane.open_current()
elif key == " ":
pane.toggle_mark()
pane.move(1)
elif key == "*":
for entry in pane.entries:
if entry.is_up:
continue
if entry.name in pane.marks:
pane.marks.discard(entry.name)
else:
pane.marks.add(entry.name)
elif key in ("a", "A"):
pane.marks = {e.name for e in pane.entries if not e.is_up}
elif key in ("u", "U"):
pane.marks.clear()
elif key in (curses.KEY_F5, "c", "C"):
self._copy(win)
elif key in (curses.KEY_F3, "v", "V"):
self._view(win, pane.current)
elif key in (curses.KEY_F7, "n", "N"):
self._mkdir(win)
elif key in (curses.KEY_F2, "m", "M"):
self._switch_mount(win)
elif key in (curses.KEY_F6, "g", "G"):
self._goto(win)
elif key in ("r", "R", curses.KEY_F9):
for item in self.panes:
item.reload()
elif key in ("h", "?", curses.KEY_F1):
self._help(win)
elif key in (curses.KEY_F10, "q", "Q"):
return
elif is_escape(key):
if confirm(win, "Explorer beenden?"):
return
# -- Aktionen ---------------------------------------------------------
def _copy(self, win):
source, target = self.source, self.target
if target.readonly:
message(win, "Das Ziel ist schreibgeschuetzt - ein Snapshot laesst sich "
"nicht beschreiben.", error=True)
return
entries = source.selection()
if not entries:
message(win, "Nichts ausgewaehlt.", error=True)
return
def progress(name, done, total):
height, width = win.getmaxyx()
share = (done / total * 100) if total else 100
bar_width = max(10, width - 46)
filled = int(bar_width * share / 100)
fill(win, height - 2,
" %s [%s%s] %3d%% %s / %s (ESC bricht ab)"
% (truncate(name, 24).ljust(24), "#" * filled,
"." * (bar_width - filled), share,
human_bytes(done), human_bytes(total)),
curses.color_pair(C_SEL))
win.refresh()
copier = Copier(win, progress)
progress("Groesse ermitteln ...", 0, 0)
total = copier.measure(entries)
free = target.free_space()
if free and total > free:
if not confirm(win, "Es werden %s benoetigt, frei sind nur %s - trotzdem?"
% (human_bytes(total), human_bytes(free))):
return
what = ("%d Eintraege" % len(entries)) if len(entries) > 1 else entries[0].name
if not confirm(win, "%s (%s) nach %s kopieren?"
% (truncate(what, 30), human_bytes(total), target.path)):
return
started = time.time()
try:
copier.copy(entries, target.path)
summary = ("%d Objekt(e) kopiert, %s in %.1fs"
% (copier.copied, human_bytes(copier.done), time.time() - started))
except Aborted:
summary = "Abgebrochen - %d Objekt(e) waren fertig" % copier.copied
except SnapfsError as exc:
summary = str(exc)
source.marks.clear()
target.reload()
if copier.failed or copier.skipped:
self._report(win, summary, copier)
else:
message(win, summary)
def _report(self, win, summary, copier):
height, width = win.getmaxyx()
lines = [summary, ""]
if copier.failed:
lines.append("Fehlgeschlagen (%d):" % len(copier.failed))
lines += [" " + item for item in copier.failed[:10]]
if copier.skipped:
lines.append("Uebersprungen (%d):" % len(copier.skipped))
lines += [" " + item for item in copier.skipped[:10]]
win.erase()
fill(win, 0, " Ergebnis", curses.color_pair(C_HEADER) | curses.A_BOLD)
for row, line in enumerate(lines[: height - 3]):
put(win, row + 2, 2, truncate(line, width - 4))
fill(win, height - 1, " beliebige Taste = weiter", curses.color_pair(C_FOOTER))
win.refresh()
read_key(win)
def _mkdir(self, win):
pane = self.source
if pane.readonly:
message(win, "Im Snapshot laesst sich nichts anlegen.", error=True)
return
name = prompt(win, "Neues Verzeichnis", "")
if not name:
return
try:
os.makedirs(os.path.join(pane.path, name), exist_ok=False)
pane.reload()
except OSError as exc:
message(win, "Nicht angelegt: %s" % (exc.strerror or exc), error=True)
def _goto(self, win):
pane = self.source
path = prompt(win, "Verzeichnis wechseln nach", pane.path)
if not path:
return
if not pane.go(os.path.expanduser(path)):
message(win, "Kein Verzeichnis: %s" % path, error=True)
def _switch_mount(self, win):
if not self.session or len(self.session.mounts) < 2:
message(win, "Es ist nur ein Dateisystem eingebunden.")
return
entries = []
for device, mountpoint in sorted(self.session.mounts.items()):
entries.append((mountpoint, "%s -> %s" % (device, mountpoint)))
chosen = choose(win, "Welches Dateisystem?", entries)
if chosen:
self.panes[0].root = chosen
self.panes[0].go(chosen)
def _view(self, win, entry):
if not entry.name or entry.is_dir or entry.is_special:
return
try:
with open(entry.path, "rb") as handle:
raw = handle.read(1024 * 1024)
except OSError as exc:
message(win, "Nicht lesbar: %s" % (exc.strerror or exc), error=True)
return
binary = b"\0" in raw[:8192]
if binary:
lines = _hexdump(raw[: 64 * 1024])
else:
lines = raw.decode("utf-8", "replace").splitlines() or ["(leer)"]
top = 0
while True:
height, width = win.getmaxyx()
rows = height - 3
win.erase()
fill(win, 0, " %s %s%s" % (truncate(entry.path, width - 30),
human_bytes(entry.size),
" [binaer, Hex-Ansicht]" if binary else ""),
curses.color_pair(C_HEADER) | curses.A_BOLD)
for row in range(rows):
position = top + row
if position >= len(lines):
break
put(win, row + 1, 0, truncate(lines[position].rstrip("\n"), width - 1))
fill(win, height - 1,
" Pfeiltasten/Bild blaettern | Pos1/Ende | q zurueck (Zeile %d von %d)"
% (min(top + 1, len(lines)), len(lines)), curses.color_pair(C_FOOTER))
win.refresh()
key = read_key(win)
if key in (curses.KEY_DOWN, "j"):
top += 1
elif key in (curses.KEY_UP, "k"):
top -= 1
elif key == curses.KEY_NPAGE or key == " ":
top += rows
elif key == curses.KEY_PPAGE:
top -= rows
elif key == curses.KEY_HOME:
top = 0
elif key == curses.KEY_END:
top = len(lines) - rows
elif key in ("q", "Q", curses.KEY_F3, curses.KEY_F10) or is_escape(key):
return
top = max(0, min(top, max(0, len(lines) - 1)))
def _help(self, win):
lines = [
"Tab zwischen den Fenstern wechseln",
"Pfeiltasten bewegen, Enter oeffnet Verzeichnis oder zeigt Datei",
"Backspace ein Verzeichnis hoch",
"Leertaste markieren, * umkehren, a alle, u keine",
"F5 / c markiertes ins andere Fenster kopieren",
"F3 / v Datei ansehen (Text oder Hex)",
"F7 / n neues Verzeichnis (nur im lokalen Fenster)",
"F6 / g Verzeichnis direkt eingeben",
"F2 / m anderes Dateisystem des Snapshots waehlen",
"r neu einlesen",
"F10 / q beenden",
"",
"Der Snapshot ist schreibgeschuetzt eingebunden - es kann nichts",
"kaputtgehen. Kopiert wird immer vom aktiven ins andere Fenster.",
"Geraetedateien und Sockets werden uebersprungen, symbolische",
"Verweise bleiben Verweise.",
]
height, width = win.getmaxyx()
win.erase()
fill(win, 0, " Tastenbelegung", curses.color_pair(C_HEADER) | curses.A_BOLD)
for row, line in enumerate(lines[: height - 3]):
put(win, row + 2, 3, truncate(line, width - 5))
fill(win, height - 1, " beliebige Taste = zurueck", curses.color_pair(C_FOOTER))
win.refresh()
read_key(win)
def _hexdump(raw, width=16):
lines = []
for offset in range(0, len(raw), width):
chunk = raw[offset:offset + width]
hexpart = " ".join("%02x" % byte for byte in chunk).ljust(width * 3)
text = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk)
lines.append("%08x %s |%s|" % (offset, hexpart, text))
return lines
# ---------------------------------------------------------------------------
# Auswahl von Gast und Snapshot
# ---------------------------------------------------------------------------
def _pick_guest(win, guests):
entries = [(g, "%s %-6d %-24s %-8s %s"
% ("LXC" if g.type == "lxc" else "VM ", g.vmid,
truncate(g.name or "-", 24), g.status, g.node))
for g in guests]
return choose(win, "Welcher Gast?", entries)
def _pick_snapshot(win, snapshots):
entries = []
for snap in snapshots:
when = (datetime.fromtimestamp(snap.snaptime).strftime("%d.%m.%Y %H:%M")
if snap.snaptime else "?")
entries.append((snap.name, "%-34s %s %s"
% (truncate(snap.name, 34), when,
truncate(snap.description.replace("\n", " "), 40))))
return choose(win, "Welcher Snapshot?", entries)
def _wait_screen(win, text):
height, width = win.getmaxyx()
win.erase()
fill(win, 0, " pvesnap-explorer", curses.color_pair(C_HEADER) | curses.A_BOLD)
put(win, height // 2, max(2, (width - len(text)) // 2), text, curses.A_BOLD)
win.refresh()
def _show_notes(win, notes, title="Hinweise"):
if not notes:
return
height, width = win.getmaxyx()
win.erase()
fill(win, 0, " %s" % title, curses.color_pair(C_HEADER) | curses.A_BOLD)
for row, note in enumerate(notes[: height - 3]):
put(win, row + 2, 2, truncate("- " + note, width - 4), curses.color_pair(C_WARN))
fill(win, height - 1, " beliebige Taste = weiter", curses.color_pair(C_FOOTER))
win.refresh()
read_key(win)
# ---------------------------------------------------------------------------
# Ablauf
# ---------------------------------------------------------------------------
def _main(stdscr, args):
curses.curs_set(0)
if hasattr(curses, "set_escdelay"):
curses.set_escdelay(25)
init_colors()
stdscr.keypad(True)
local = Pane(path=os.path.realpath(args.local), root="/", readonly=False,
title="Lokaler Rechner")
if args.path:
left = Pane(path=os.path.realpath(args.path), root=os.path.realpath(args.path),
readonly=True, title="Verzeichnis: %s" % args.path)
Explorer(left, local, subtitle=args.path).run(stdscr)
return 0
_wait_screen(stdscr, "Frage Proxmox nach VMs und Containern ...")
proxmox = Proxmox()
guests = proxmox.inventory()
if not guests:
message(stdscr, "Keine VMs oder Container gefunden.", error=True)
return 1
if args.vmid:
guest = next((g for g in guests if g.vmid == args.vmid), None)
if guest is None:
message(stdscr, "Gast %d nicht gefunden." % args.vmid, error=True)
return 2
else:
guest = _pick_guest(stdscr, guests)
if guest is None:
return 0
_wait_screen(stdscr, "Lese Snapshots von %s ..." % guest.label)
snapshots = list_snapshots(proxmox, guest)
if not snapshots:
message(stdscr, "%s hat keine Snapshots." % guest.label, error=True)
return 1
if args.snapshot:
name = next((s.name for s in snapshots if s.name == args.snapshot), None)
if name is None:
message(stdscr, "Snapshot %r nicht gefunden." % args.snapshot, error=True)
return 2
else:
name = _pick_snapshot(stdscr, snapshots)
if name is None:
return 0
session = Session(proxmox, guest, name)
try:
_wait_screen(stdscr, "Binde Snapshot '%s' ein - das kann etwas dauern ..." % name)
session.open()
session.prepare()
except (SnapfsError, ProxmoxError) as exc:
session.close()
_show_notes(stdscr, str(exc).splitlines() + session.notes, "Fehlgeschlagen")
return 3
try:
_show_notes(stdscr, session.notes)
mountpoints = sorted(set(session.mounts.values()))
start = mountpoints[0]
if len(mountpoints) > 1:
entries = [(m, m) for m in mountpoints]
start = choose(stdscr, "Welches Dateisystem zuerst?", entries) or mountpoints[0]
left = Pane(path=start, root=start, readonly=True,
title="%s @ %s" % (guest.label, name))
Explorer(left, local, session=session,
subtitle="%s - Snapshot %s" % (guest.label, name)).run(stdscr)
finally:
_wait_screen(stdscr, "Haenge Snapshot wieder aus ...")
session.close()
return 0
def build_parser():
parser = argparse.ArgumentParser(
prog="pvesnap-explorer",
description="Dateien aus einem Proxmox-Snapshot holen (zwei Fenster wie im "
"Midnight Commander).")
parser.add_argument("vmid", nargs="?", type=int, help="VMID des Gastes")
parser.add_argument("snapshot", nargs="?", help="Name des Snapshots")
parser.add_argument("-l", "--local", default=os.getcwd(),
help="Startverzeichnis im rechten Fenster (Vorgabe: aktuelles)")
parser.add_argument("-p", "--path",
help="beliebiges Verzeichnis oeffnen, ohne Proxmox zu fragen")
parser.add_argument("--cleanup", action="store_true",
help="haengengebliebene Einbindungen aufraeumen und beenden")
parser.add_argument("-v", "--verbose", action="store_true")
parser.add_argument("-V", "--version", action="version",
version="pvesnap-explorer %s" % __version__)
return parser
def main(argv=None):
args = build_parser().parse_args(argv)
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
format="%(levelname)s %(message)s")
if args.cleanup:
removed = cleanup_leftovers(verbose=True)
print("%d Eintrag/Eintraege aufgeraeumt." % len(removed))
return 0
if not args.path:
if not pvesh_available():
print("'pvesh' nicht gefunden - der Explorer laeuft auf dem Proxmox-Host.\n"
"Zum Ausprobieren: pvesnap-explorer --path /ein/verzeichnis",
file=sys.stderr)
return 3
if os.geteuid() != 0:
print("Der Explorer braucht root-Rechte (Snapshots einbinden und mounten).",
file=sys.stderr)
return 4
# Waehrend der Oberflaeche darf nichts ins Terminal protokolliert werden.
handlers = logging.getLogger().handlers[:]
if not args.verbose:
logging.getLogger().handlers = []
try:
return curses.wrapper(_main, args)
except KeyboardInterrupt:
return 130
finally:
logging.getLogger().handlers = handlers
if __name__ == "__main__": # pragma: no cover
sys.exit(main())
+592
View File
@@ -0,0 +1,592 @@
"""Snapshots als Dateisystem zugaenglich machen.
Ein Proxmox-Snapshot ist kein Verzeichnis - je nach Storage muss er erst
zugaenglich gemacht werden:
rbd (Ceph) rbd map pool/image@snap -> /dev/rbdN
zfspool zfs clone ds@snap tmp -> /dev/zvol/tmp
(bei Container-Datasets direkt .zfs/snapshot/<snap>)
lvmthin lvchange -ay vg/snap_vol_snap -> /dev/vg/snap_...
dir/nfs qemu-nbd --load-snapshot -> /dev/nbdN
Danach werden die Partitionen ermittelt und **schreibgeschuetzt** gemountet.
Alles, was angelegt wurde, merkt sich diese Schicht und raeumt es wieder ab -
auch nach einem Absturz ueber `cleanup_leftovers()`.
Auf den Inhalt greifen Explorer und Web-Oberflaeche dann nur noch als
gewoehnlichen Pfad zu.
"""
from __future__ import annotations
import json
import logging
import os
import re
import shutil
import subprocess
import time
from dataclasses import dataclass, field
log = logging.getLogger("pvesnap.snapfs")
RUN_DIR = "/run/pvesnap"
MOUNT_ROOT = os.path.join(RUN_DIR, "mnt")
REGISTRY = os.path.join(RUN_DIR, "explorer.json")
# Datentraeger-Schluessel in den Gast-Konfigurationen
QEMU_DISK_KEY = re.compile(r"^(ide|sata|scsi|virtio)\d+$")
LXC_DISK_KEY = re.compile(r"^(rootfs|mp\d+)$")
# Mount-Optionen je Dateisystem. Snapshots laufender Gaeste haben fast immer
# ein unsauberes Journal - ohne diese Optionen wuerde der Kernel versuchen,
# es zurueckzuschreiben.
MOUNT_OPTIONS = {
"ext2": ["ro"],
"ext3": ["ro", "noload"],
"ext4": ["ro", "noload"],
"xfs": ["ro", "norecovery", "nouuid"],
"btrfs": ["ro", "nologreplay"],
"vfat": ["ro"],
"ntfs": ["ro"],
"ntfs3": ["ro"],
"iso9660": ["ro"],
}
DEFAULT_MOUNT_OPTIONS = ["ro"]
MOUNTABLE = set(MOUNT_OPTIONS) | {"ext4", "xfs", "btrfs"}
IGNORED_FSTYPES = {"", "swap", "linux_raid_member", "crypto_LUKS", "LVM2_member"}
class SnapfsError(Exception):
"""Ein Snapshot liess sich nicht zugaenglich machen."""
# ---------------------------------------------------------------------------
# Kommandos
# ---------------------------------------------------------------------------
def run(args, timeout=120, check=True):
"""Externes Kommando ausfuehren; gibt die Standardausgabe zurueck."""
log.debug("exec: %s", " ".join(args))
try:
proc = subprocess.run(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
timeout=timeout)
except FileNotFoundError:
raise SnapfsError("Kommando nicht gefunden: %s" % args[0])
except subprocess.TimeoutExpired:
raise SnapfsError("Zeitueberschreitung bei: %s" % " ".join(args))
stdout = proc.stdout.decode("utf-8", "replace").strip()
stderr = proc.stderr.decode("utf-8", "replace").strip()
if check and proc.returncode != 0:
raise SnapfsError("%s: %s" % (args[0], (stderr or stdout or "Fehler").splitlines()[0]))
return stdout
def have(command):
return shutil.which(command) is not None
# ---------------------------------------------------------------------------
# Datenmodell
# ---------------------------------------------------------------------------
@dataclass
class Volume:
"""Ein Datentraeger aus der Gast-Konfiguration."""
key: str # scsi0, rootfs, mp0 ...
volid: str # data:vm-802-disk-0
storage: str # data
volname: str # vm-802-disk-0
size: str = ""
@property
def label(self):
return "%s %s%s" % (self.key.ljust(8), self.volid,
(" (%s)" % self.size) if self.size else "")
@dataclass
class Device:
"""Ein Blockgeraet (ganze Platte oder Partition) des Snapshots."""
path: str
name: str = ""
size: int = 0
fstype: str = ""
label: str = ""
kind: str = "part"
volume: object = None
@property
def mountable(self):
return bool(self.fstype) and self.fstype not in IGNORED_FSTYPES
@property
def human_size(self):
return human_bytes(self.size)
@property
def title(self):
parts = [os.path.basename(self.path)]
if self.label:
parts.append('"%s"' % self.label)
parts.append(self.fstype or "unbekannt")
parts.append(self.human_size)
return " ".join(parts)
def human_bytes(count):
value = float(count or 0)
for unit in ("B", "K", "M", "G", "T", "P"):
if value < 1024 or unit == "P":
return "%d%s" % (value, unit) if unit == "B" else "%.1f%s" % (value, unit)
value /= 1024
return "%.1fP" % value
# ---------------------------------------------------------------------------
# Konfiguration des Snapshots lesen
# ---------------------------------------------------------------------------
def snapshot_config(proxmox, guest, snapname):
path = "/nodes/%s/%s/%d/snapshot/%s/config" % (guest.node, guest.type,
guest.vmid, snapname)
data = proxmox._json(["get", path])
if not isinstance(data, dict):
raise SnapfsError("Konfiguration des Snapshots %r nicht lesbar" % snapname)
return data
def volumes_from_config(config, guest_type):
"""Alle echten Datentraeger aus einer Gast-Konfiguration."""
pattern = LXC_DISK_KEY if guest_type == "lxc" else QEMU_DISK_KEY
volumes = []
for key in sorted(config):
if not pattern.match(key):
continue
value = str(config[key])
if not value or "media=cdrom" in value:
continue
volid = value.split(",", 1)[0].strip()
if volid in ("none", "cdrom") or ":" not in volid:
continue
storage, volname = volid.split(":", 1)
size = ""
for part in value.split(",")[1:]:
if part.strip().startswith("size="):
size = part.strip()[5:]
volumes.append(Volume(key=key, volid=volid, storage=storage,
volname=volname, size=size))
return volumes
def storage_info(proxmox, name):
data = proxmox._json(["get", "/storage/%s" % name])
if not isinstance(data, dict):
raise SnapfsError("Storage %r nicht gefunden" % name)
return data
# ---------------------------------------------------------------------------
# Merkliste, damit nach einem Absturz nichts stehen bleibt
# ---------------------------------------------------------------------------
def _registry_load():
try:
with open(REGISTRY, "r", encoding="utf-8") as handle:
data = json.load(handle)
return data if isinstance(data, list) else []
except (OSError, ValueError):
return []
def _registry_save(entries):
try:
os.makedirs(RUN_DIR, exist_ok=True)
tmp = REGISTRY + ".tmp"
with open(tmp, "w", encoding="utf-8") as handle:
json.dump(entries, handle, indent=1)
os.replace(tmp, REGISTRY)
except OSError as exc:
log.debug("Merkliste nicht schreibbar: %s", exc)
def _registry_add(kind, value, extra=None):
entries = _registry_load()
entries.append({"kind": kind, "value": value, "extra": extra or {},
"pid": os.getpid(), "time": int(time.time())})
_registry_save(entries)
def _registry_remove(kind, value):
entries = [e for e in _registry_load()
if not (e.get("kind") == kind and e.get("value") == value)]
_registry_save(entries)
def cleanup_leftovers(verbose=False):
"""Raeumt auf, was ein abgestuerzter Lauf hinterlassen hat."""
removed = []
for entry in reversed(_registry_load()):
kind, value = entry.get("kind"), entry.get("value")
extra = entry.get("extra") or {}
try:
if kind == "mount":
_umount(value)
elif kind == "rbd":
run(["rbd", "unmap", value], check=False)
elif kind == "rbd-nbd":
run(["rbd-nbd", "unmap", value], check=False)
elif kind == "nbd":
run(["qemu-nbd", "--disconnect", value], check=False)
elif kind == "lvm":
run(["lvchange", "-an", value], check=False)
elif kind == "zfsclone":
run(["zfs", "destroy", value], check=False)
elif kind == "dir":
if os.path.isdir(value):
os.rmdir(value)
else:
continue
removed.append("%s %s" % (kind, value))
if verbose:
print("aufgeraeumt: %s %s" % (kind, value))
except (SnapfsError, OSError) as exc:
if verbose:
print("konnte %s %s nicht aufraeumen: %s" % (kind, value, exc))
_registry_remove(kind, value)
del extra
return removed
def _umount(mountpoint):
if not os.path.ismount(mountpoint):
return
try:
run(["umount", mountpoint])
except SnapfsError:
run(["umount", "-l", mountpoint], check=False) # notfalls verzoegert
# ---------------------------------------------------------------------------
# Sitzung
# ---------------------------------------------------------------------------
class Session:
"""Ein geoeffneter Snapshot: eingebundene Geraete und Mountpunkte."""
def __init__(self, proxmox, guest, snapname):
self.proxmox = proxmox
self.guest = guest
self.snapname = snapname
self.volumes = []
self.devices = []
self.mounts = {} # Geraetepfad -> Mountpunkt
self.notes = [] # Hinweise fuer die Oberflaeche
self._cleanup = [] # (art, wert) in umgekehrter Reihenfolge
# -- oeffnen ----------------------------------------------------------
def open(self):
config = snapshot_config(self.proxmox, self.guest, self.snapname)
self.volumes = volumes_from_config(config, self.guest.type)
if not self.volumes:
raise SnapfsError("Der Snapshot enthaelt keine Datentraeger.")
return self.volumes
def prepare(self):
"""Alle Datentraeger einbinden und mountbare Dateisysteme mounten."""
for volume in self.volumes:
try:
self._prepare_volume(volume)
except SnapfsError as exc:
self.notes.append("%s: %s" % (volume.key, exc))
log.warning("%s (%s): %s", volume.key, volume.volid, exc)
if not self.mounts and not self.devices:
raise SnapfsError("Kein Datentraeger des Snapshots konnte "
"eingebunden werden.\n- " + "\n- ".join(self.notes))
return self.mounts
def _prepare_volume(self, volume):
info = storage_info(self.proxmox, volume.storage)
kind = info.get("type")
if kind == "zfspool" and self.guest.type == "lxc":
# Container liegen auf ZFS als Dateisystem - der Snapshot ist ohne
# Umweg unter .zfs/snapshot/<name> sichtbar.
path = self._zfs_snapshot_dir(info, volume)
if path:
self.mounts[path] = path
self.notes.append("%s: direkt aus %s" % (volume.key, path))
return
device_path = self._map(info, volume)
for device in self._scan(device_path, volume):
self.devices.append(device)
if device.mountable:
try:
self.mount(device)
except SnapfsError as exc:
self.notes.append("%s: %s" % (device.path, exc))
# -- Einbinden je Storage-Typ -----------------------------------------
def _map(self, info, volume):
kind = info.get("type")
if kind == "rbd":
return self._map_rbd(info, volume)
if kind == "zfspool":
return self._map_zfs(info, volume)
if kind in ("lvmthin", "lvm"):
return self._map_lvm(info, volume)
if kind in ("dir", "nfs", "cifs", "glusterfs"):
return self._map_qcow(info, volume)
raise SnapfsError("Storage-Typ %r wird vom Explorer nicht unterstuetzt" % kind)
def _rbd_args(self, info):
args = []
storage = info.get("storage") or info.get("name") or ""
keyring = "/etc/pve/priv/ceph/%s.keyring" % storage
if info.get("monhost"):
args += ["-m", str(info["monhost"]).replace(" ", ",")]
args += ["--id", info.get("username") or "admin"]
if os.path.exists(keyring):
args += ["--keyring", keyring]
else:
args += ["--id", info.get("username") or "admin"]
return args
def _map_rbd(self, info, volume):
if not have("rbd"):
raise SnapfsError("'rbd' ist nicht installiert")
pool = info.get("pool") or "rbd"
namespace = info.get("namespace")
image = "%s/%s%s@%s" % (pool, (namespace + "/") if namespace else "",
volume.volname, self.snapname)
base = self._rbd_args(info)
try:
output = run(["rbd"] + base + ["map", "--read-only", image])
device = self._first_device(output)
self._remember("rbd", device)
return device
except SnapfsError as exc:
# krbd kann manche Image-Features nicht - rbd-nbd kann alle.
if not have("rbd-nbd"):
raise SnapfsError("%s (rbd-nbd waere die Alternative, ist aber "
"nicht installiert: apt install ceph-common)" % exc)
log.info("rbd map fehlgeschlagen (%s) - versuche rbd-nbd", exc)
output = run(["rbd-nbd"] + base + ["map", "--read-only", image], timeout=180)
device = self._first_device(output)
self._remember("rbd-nbd", device)
return device
def _zfs_dataset(self, info, volume):
pool = info.get("pool") or info.get("zfspool") or ""
return "%s/%s" % (pool.rstrip("/"), volume.volname)
def _zfs_snapshot_dir(self, info, volume):
dataset = self._zfs_dataset(info, volume)
try:
mountpoint = run(["zfs", "get", "-H", "-o", "value", "mountpoint", dataset])
except SnapfsError:
return ""
if not mountpoint or mountpoint in ("none", "-", "legacy"):
return ""
candidate = os.path.join(mountpoint, ".zfs", "snapshot", self.snapname)
return candidate if os.path.isdir(candidate) else ""
def _map_zfs(self, info, volume):
if not have("zfs"):
raise SnapfsError("'zfs' ist nicht installiert")
dataset = self._zfs_dataset(info, volume)
source = "%s@%s" % (dataset, self.snapname)
clone = "%s-pvesnap-%d" % (dataset, os.getpid())
run(["zfs", "clone", "-o", "readonly=on", source, clone])
self._remember("zfsclone", clone)
device = "/dev/zvol/%s" % clone
for _ in range(50): # udev braucht einen Moment
if os.path.exists(device):
return device
time.sleep(0.1)
raise SnapfsError("Klon %s ist nicht als Geraet aufgetaucht" % clone)
def _map_lvm(self, info, volume):
if not have("lvchange"):
raise SnapfsError("LVM-Werkzeuge sind nicht installiert")
vg = info.get("vgname") or ""
# Proxmox benennt Snapshot-LVs immer nach diesem Schema.
lv = "snap_%s_%s" % (volume.volname, self.snapname)
target = "%s/%s" % (vg, lv)
run(["lvchange", "-ay", "-K", "--readonly", target])
self._remember("lvm", target)
device = "/dev/%s/%s" % (vg, lv)
if not os.path.exists(device):
raise SnapfsError("%s ist nach dem Aktivieren nicht vorhanden" % device)
return device
def _map_qcow(self, info, volume):
if not have("qemu-nbd"):
raise SnapfsError("'qemu-nbd' ist nicht installiert (apt install qemu-utils)")
base = info.get("path") or ""
image = os.path.join(base, "images", str(self.guest.vmid), volume.volname)
if not os.path.exists(image):
raise SnapfsError("Abbilddatei nicht gefunden: %s" % image)
if not image.endswith(".qcow2"):
raise SnapfsError("nur qcow2 kann Snapshots enthalten (%s)"
% os.path.basename(image))
run(["modprobe", "nbd", "max_part=16"], check=False)
device = self._free_nbd()
run(["qemu-nbd", "--read-only", "--load-snapshot=%s" % self.snapname,
"--connect=%s" % device, image], timeout=180)
self._remember("nbd", device)
time.sleep(0.5)
return device
@staticmethod
def _free_nbd():
for index in range(16):
device = "/dev/nbd%d" % index
if not os.path.exists(device):
continue
try:
with open("/sys/block/nbd%d/size" % index) as handle:
if handle.read().strip() == "0":
return device
except OSError:
continue
raise SnapfsError("kein freies /dev/nbdN vorhanden")
@staticmethod
def _first_device(output):
for line in (output or "").splitlines():
line = line.strip()
if line.startswith("/dev/"):
return line
raise SnapfsError("Antwort ohne Geraetenamen: %s" % (output or "(leer)"))
# -- Partitionen und Dateisysteme -------------------------------------
def _scan(self, device_path, volume):
"""Partitionen des eingebundenen Geraets ermitteln."""
devices = self._lsblk(device_path, volume)
if len(devices) == 1 and not devices[0].mountable:
# Partitionstabelle vorhanden, aber noch nicht eingelesen
for tool in (["partx", "-a", device_path], ["kpartx", "-a", "-r", device_path]):
if have(tool[0]):
run(tool, check=False)
time.sleep(0.5)
devices = self._lsblk(device_path, volume)
if len(devices) > 1:
break
return devices
def _lsblk(self, device_path, volume):
try:
output = run(["lsblk", "-J", "-b", "-o",
"NAME,PATH,SIZE,FSTYPE,LABEL,TYPE", device_path])
data = json.loads(output)
except (SnapfsError, ValueError) as exc:
raise SnapfsError("Partitionen nicht lesbar: %s" % exc)
found = []
def walk(entries):
for entry in entries:
device = Device(
path=entry.get("path") or "",
name=entry.get("name") or "",
size=int(entry.get("size") or 0),
fstype=entry.get("fstype") or "",
label=entry.get("label") or "",
kind=entry.get("type") or "part",
volume=volume,
)
children = entry.get("children") or []
if device.path:
# Eine Platte mit Partitionen selbst nicht anbieten
if not (children and not device.fstype):
found.append(device)
if device.fstype == "LVM2_member":
self.notes.append(
"%s enthaelt LVM - im Gast angelegte Volume-Groups werden "
"nicht automatisch aktiviert (Namenskonflikte mit dem Host)"
% device.path)
walk(children)
walk(data.get("blockdevices") or [])
return found
def mount(self, device):
if device.path in self.mounts:
return self.mounts[device.path]
name = "%s-%d-%s-%s" % (self.guest.type, self.guest.vmid,
re.sub(r"[^A-Za-z0-9_.-]", "_", self.snapname),
os.path.basename(device.path))
mountpoint = os.path.join(MOUNT_ROOT, name)
os.makedirs(mountpoint, exist_ok=True)
self._remember("dir", mountpoint)
options = MOUNT_OPTIONS.get(device.fstype, DEFAULT_MOUNT_OPTIONS)
attempts = [options]
if options != DEFAULT_MOUNT_OPTIONS:
attempts.append(DEFAULT_MOUNT_OPTIONS) # notfalls schlicht "ro"
last = None
for option_set in attempts:
try:
run(["mount", "-o", ",".join(option_set), "-t", device.fstype or "auto",
device.path, mountpoint])
self.mounts[device.path] = mountpoint
self._remember("mount", mountpoint)
log.info("%s (%s) eingehaengt unter %s",
device.path, device.fstype, mountpoint)
return mountpoint
except SnapfsError as exc:
last = exc
raise SnapfsError("%s nicht einhaengbar: %s" % (device.path, last))
# -- aufraeumen -------------------------------------------------------
def _remember(self, kind, value):
self._cleanup.append((kind, value))
_registry_add(kind, value)
def close(self):
for kind, value in reversed(self._cleanup):
try:
if kind == "mount":
_umount(value)
elif kind == "rbd":
run(["rbd", "unmap", value], check=False)
elif kind == "rbd-nbd":
run(["rbd-nbd", "unmap", value], check=False)
elif kind == "nbd":
run(["qemu-nbd", "--disconnect", value], check=False)
elif kind == "lvm":
run(["lvchange", "-an", value], check=False)
elif kind == "zfsclone":
run(["zfs", "destroy", value], check=False)
elif kind == "dir" and os.path.isdir(value):
os.rmdir(value)
except (SnapfsError, OSError) as exc:
log.warning("Aufraeumen von %s %s fehlgeschlagen: %s", kind, value, exc)
_registry_remove(kind, value)
self._cleanup = []
self.mounts = {}
def __enter__(self):
return self
def __exit__(self, *_exc):
self.close()
return False
def list_snapshots(proxmox, guest):
"""Snapshots eines Gastes, neueste zuerst."""
snapshots = proxmox.list_snapshots(guest)
return sorted(snapshots, key=lambda s: s.snaptime, reverse=True)
+45 -205
View File
@@ -9,6 +9,9 @@ import curses
import os
import subprocess
from .curses_util import (C_DIM, C_FOOTER, C_HEADER, C_OK, C_SEL, C_WARN,
choose, confirm, fill, fill_row, init_colors, is_enter,
is_escape, keybar, message, prompt, put, read_key)
from .config import (WEEKDAY_NAMES, Config, ConfigError, Group, clone_group,
load_config, parse_time_of_day, save_config)
from .naming import effective_slug
@@ -32,8 +35,6 @@ TYPE_CHOICES = [
(["lxc"], "nur Container (LXC)"),
]
C_HEADER, C_FOOTER, C_SEL, C_WARN, C_OK, C_DIM = 1, 2, 3, 4, 5, 6
SERVICE = "pvesnap.service"
# Tastenleiste der Gruppenliste: (Taste, lange Beschriftung, kurze Beschriftung)
@@ -93,80 +94,6 @@ def service_text():
"activating": "Dienst: startet"}.get(state, "Dienst: %s" % state)
def _keybar(width, items):
"""Tastenleiste als Liste von Zeilen, passend zur Breite des Terminals.
Reicht der Platz nicht fuer eine Zeile, wird auf zwei Zeilen umgebrochen;
bei ganz schmalen Terminals bleiben nur noch die Tasten uebrig.
"""
def join(entries, index):
return " | ".join("%s %s" % (entry[0], entry[index]) for entry in entries)
long_text = join(items, 1)
if len(long_text) + 2 <= width:
return [long_text]
half = (len(items) + 1) // 2
first, second = join(items[:half], 1), join(items[half:], 1)
if max(len(first), len(second)) + 2 <= width:
return [first, second]
short_text = join(items, 2)
if len(short_text) + 2 <= width:
return [short_text]
first, second = join(items[:half], 2), join(items[half:], 2)
if max(len(first), len(second)) + 2 <= width:
return [first, second]
return [" ".join(entry[0] for entry in items)]
# ---------------------------------------------------------------------------
# Zeichen-Helfer
# ---------------------------------------------------------------------------
def _put(win, y, x, text, attr=0):
"""addstr, das bei zu kleinem Terminal nicht abstuerzt."""
height, width = win.getmaxyx()
if y < 0 or y >= height or x >= width:
return
try:
win.addstr(y, x, str(text)[: max(0, width - x - 1)], attr)
except curses.error:
pass
def _fill(win, y, text, attr):
height, width = win.getmaxyx()
if y < 0 or y >= height:
return
try:
win.addstr(y, 0, str(text).ljust(width - 1)[: width - 1], attr)
except curses.error:
pass
def _read_key(win):
try:
return win.get_wch()
except curses.error:
return None
except KeyboardInterrupt:
return "\x1b"
def _is_enter(key):
return key in ("\n", "\r", curses.KEY_ENTER)
def _is_escape(key):
return key == "\x1b"
def _is_backspace(key):
return key in ("\x7f", "\b", "\x08", curses.KEY_BACKSPACE)
class Editor:
def __init__(self, path):
self.path = path
@@ -206,99 +133,25 @@ class Editor:
def _frame(self, win, title, subtitle, keys):
win.erase()
height, width = win.getmaxyx()
_fill(win, 0, " pvesnap - %s" % title, curses.color_pair(C_HEADER) | curses.A_BOLD)
fill(win, 0, " pvesnap - %s" % title, curses.color_pair(C_HEADER) | curses.A_BOLD)
marker = "*" if self.dirty else " "
_fill(win, 1, " %s%s" % (marker, subtitle), curses.color_pair(C_DIM))
fill(win, 1, " %s%s" % (marker, subtitle), curses.color_pair(C_DIM))
lines = list(keys) if isinstance(keys, (list, tuple)) else [keys]
for offset, line in enumerate(reversed(lines)):
_fill(win, height - 1 - offset, " " + line, curses.color_pair(C_FOOTER))
fill(win, height - 1 - offset, " " + line, curses.color_pair(C_FOOTER))
return height, width
def _message(self, win, text, error=False):
height, _ = win.getmaxyx()
attr = curses.color_pair(C_WARN if error else C_OK) | curses.A_BOLD
_fill(win, height - 2, " " + str(text) + " [Taste druecken]", attr)
win.refresh()
_read_key(win)
return message(win, text, error)
def _confirm(self, win, question):
height, _ = win.getmaxyx()
_fill(win, height - 2, " %s [j/n]" % question,
curses.color_pair(C_WARN) | curses.A_BOLD)
win.refresh()
while True:
key = _read_key(win)
if key in ("j", "J", "y", "Y"):
return True
if key in ("n", "N") or _is_escape(key):
return False
return confirm(win, question)
def _prompt(self, win, label, initial=""):
"""Einzeilige Eingabe am unteren Rand. None = abgebrochen."""
height, width = win.getmaxyx()
buffer = list(str(initial))
curses.curs_set(1)
try:
while True:
text = "".join(buffer)
prompt = " %s: " % label
visible = width - len(prompt) - 2
shown = text[-visible:] if visible > 0 and len(text) > visible else text
_fill(win, height - 2, prompt + shown, curses.color_pair(C_SEL))
try:
win.move(height - 2, min(width - 2, len(prompt) + len(shown)))
except curses.error:
pass
win.refresh()
key = _read_key(win)
if key is None:
continue
if _is_enter(key):
return "".join(buffer).strip()
if _is_escape(key):
return None
if _is_backspace(key):
if buffer:
buffer.pop()
continue
if key == "\x15": # Strg-U
buffer = []
continue
if isinstance(key, str) and key.isprintable():
buffer.append(key)
finally:
curses.curs_set(0)
return prompt(win, label, initial)
def _choose(self, win, title, entries, current=0):
"""Kleines Auswahlfenster. entries: Liste von (wert, beschriftung)."""
height, width = win.getmaxyx()
box_h = min(len(entries) + 4, height - 2)
box_w = min(max([len(title)] + [len(e[1]) for e in entries]) + 8, width - 2)
top = max(0, (height - box_h) // 2)
left = max(0, (width - box_w) // 2)
box = curses.newwin(box_h, box_w, top, left)
box.keypad(True)
index = max(0, min(current, len(entries) - 1))
while True:
box.erase()
box.box()
_put(box, 0, 2, " %s " % title, curses.A_BOLD)
for row, (_value, label) in enumerate(entries[: box_h - 4]):
attr = curses.color_pair(C_SEL) if row == index else 0
_put(box, row + 2, 2, " " + label.ljust(box_w - 5), attr)
box.refresh()
key = _read_key(box)
if key in (curses.KEY_UP, "k"):
index = (index - 1) % len(entries)
elif key in (curses.KEY_DOWN, "j"):
index = (index + 1) % len(entries)
elif _is_enter(key):
return entries[index][0]
elif _is_escape(key) or key == "q":
return None
return choose(win, title, entries, current)
# -- Hauptschleife ----------------------------------------------------
@@ -306,17 +159,7 @@ class Editor:
curses.curs_set(0)
if hasattr(curses, "set_escdelay"):
curses.set_escdelay(25)
try:
curses.start_color()
curses.use_default_colors()
curses.init_pair(C_HEADER, curses.COLOR_WHITE, curses.COLOR_BLUE)
curses.init_pair(C_FOOTER, curses.COLOR_BLACK, curses.COLOR_WHITE)
curses.init_pair(C_SEL, curses.COLOR_BLACK, curses.COLOR_CYAN)
curses.init_pair(C_WARN, curses.COLOR_WHITE, curses.COLOR_RED)
curses.init_pair(C_OK, curses.COLOR_BLACK, curses.COLOR_GREEN)
curses.init_pair(C_DIM, -1, -1)
except curses.error:
pass
init_colors()
stdscr.keypad(True)
self.screen_groups(stdscr)
@@ -336,16 +179,16 @@ class Editor:
else "" + self.path[-(room - 1):])
height, width = self._frame(
win, "Gruppen", "Datei: %s [%s]" % (path_text, status),
_keybar(width_now, GROUP_KEYS))
keybar(width_now, GROUP_KEYS))
_put(win, 3, 2, "Gruppe".ljust(18) + "Aktiv " + "Zeitplan".ljust(30)
put(win, 3, 2, "Gruppe".ljust(18) + "Aktiv " + "Zeitplan".ljust(30)
+ "Behalte".ljust(20) + "Auswahl", curses.A_BOLD)
_put(win, 4, 2, "-" * max(0, width - 4), curses.color_pair(C_DIM))
put(win, 4, 2, "-" * max(0, width - 4), curses.color_pair(C_DIM))
if not groups:
_put(win, 6, 4, "Noch keine Gruppe angelegt - mit 'n' eine erste anlegen.")
_put(win, 8, 4, "Beispiel: 'stuendlich' alle 1h, 24 Snapshots behalten;")
_put(win, 9, 4, " 'taeglich' um 02:30, 14 Tage Vorhaltezeit.")
put(win, 6, 4, "Noch keine Gruppe angelegt - mit 'n' eine erste anlegen.")
put(win, 8, 4, "Beispiel: 'stuendlich' alle 1h, 24 Snapshots behalten;")
put(win, 9, 4, " 'taeglich' um 02:30, 14 Tage Vorhaltezeit.")
visible = max(1, height - 8)
start = max(0, min(index - visible // 2, max(0, len(groups) - visible)))
@@ -360,15 +203,15 @@ class Editor:
truncate(describe(group), 29).ljust(30),
truncate(keep, 19).ljust(20),
truncate(self._selection_text(group), max(4, width - 80))))
_fill_row(win, 5 + row, 2, line, width, attr)
fill_row(win, 5 + row, 2, line, width, attr)
problems = self.config.validate()
if problems:
_fill(win, height - 2, " %d Hinweis(e) - 'p' zeigt Details"
fill(win, height - 2, " %d Hinweis(e) - 'p' zeigt Details"
% len(problems), curses.color_pair(C_WARN))
win.refresh()
key = _read_key(win)
key = read_key(win)
if key in (curses.KEY_UP, "k"):
index -= 1
@@ -378,7 +221,7 @@ class Editor:
index = 0
elif key == curses.KEY_END:
index = len(groups) - 1
elif _is_enter(key) and groups:
elif is_enter(key) and groups:
self.screen_group(win, groups[index])
elif key == "n":
self._new_group(win)
@@ -405,7 +248,7 @@ class Editor:
self._save(win)
elif key in ("r", "R"):
self._reload_service(win)
elif key in ("q", "Q") or _is_escape(key):
elif key in ("q", "Q") or is_escape(key):
if self.dirty and not self._confirm(win, "Ungespeicherte Aenderungen verwerfen?"):
continue
return
@@ -441,11 +284,11 @@ class Editor:
height, width = self._frame(win, "Hinweise zur Konfiguration", self.path,
"beliebige Taste = zurueck")
if not problems:
_put(win, 4, 4, "Alles in Ordnung.", curses.color_pair(C_OK))
put(win, 4, 4, "Alles in Ordnung.", curses.color_pair(C_OK))
for row, problem in enumerate(problems[: height - 6]):
_put(win, 4 + row, 4, truncate("- " + problem, width - 6))
put(win, 4 + row, 4, truncate("- " + problem, width - 6))
win.refresh()
_read_key(win)
read_key(win)
def _save(self, win, offer_reload=True):
problems = self.config.validate()
@@ -511,7 +354,7 @@ class Editor:
def _run_service_action(self, win, action, past_tense):
height, _ = win.getmaxyx()
_fill(win, height - 2, " systemctl %s %s ..." % (action, SERVICE),
fill(win, height - 2, " systemctl %s %s ..." % (action, SERVICE),
curses.color_pair(C_SEL))
win.refresh()
code, output = _systemctl(action, SERVICE)
@@ -553,16 +396,16 @@ class Editor:
break
kind, label, value, _handler = fields[position]
if kind == "-":
_put(win, 3 + row, 2, "-- %s " % label + "-" * max(0, width - len(label) - 8),
put(win, 3 + row, 2, "-- %s " % label + "-" * max(0, width - len(label) - 8),
curses.color_pair(C_DIM) | curses.A_BOLD)
continue
attr = curses.color_pair(C_SEL) if position == index else 0
_fill_row(win, 3 + row, 2,
fill_row(win, 3 + row, 2,
truncate(label, 30).ljust(32) + truncate(value, max(4, width - 40)),
width, attr)
win.refresh()
key = _read_key(win)
key = read_key(win)
if key in (curses.KEY_UP, "k"):
index = self._step(fields, index, -1)
@@ -572,14 +415,14 @@ class Editor:
index = self._step(fields, min(index + 10, len(fields) - 1), +1, land=True)
elif key == curses.KEY_PPAGE:
index = self._step(fields, max(index - 10, 0), -1, land=True)
elif _is_enter(key) or key == " ":
elif is_enter(key) or key == " ":
handler = fields[index][3]
if handler:
handler(win)
self.dirty = True
elif key == "v":
self._pick_vms(win, group)
elif key in ("q", "Q") or _is_escape(key):
elif key in ("q", "Q") or is_escape(key):
return
@staticmethod
@@ -808,7 +651,7 @@ class Editor:
"Leer Auswaehlen | a Alle | n Keine | i Umkehren | / Filter | "
"Enter Uebernehmen | q Abbruch")
_put(win, 3, 2, " VMID Typ Name".ljust(46) + "Node".ljust(14)
put(win, 3, 2, " VMID Typ Name".ljust(46) + "Node".ljust(14)
+ "Status".ljust(10) + "Tags", curses.A_BOLD)
visible = max(1, height - 6)
if index < top:
@@ -831,10 +674,10 @@ class Editor:
truncate(guest.node, 12).ljust(14),
truncate(guest.status, 9).ljust(10),
truncate(",".join(guest.tags) or "-", 20)))
_fill_row(win, 4 + row, 2, line, width, attr)
fill_row(win, 4 + row, 2, line, width, attr)
win.refresh()
key = _read_key(win)
key = read_key(win)
if key in (curses.KEY_UP, "k"):
index -= 1
@@ -872,11 +715,11 @@ class Editor:
elif key == "r":
self.guests(refresh=True)
guests = self._guests
elif _is_enter(key):
elif is_enter(key):
group.vmids = sorted(chosen)
self.dirty = True
return
elif key in ("q", "Q") or _is_escape(key):
elif key in ("q", "Q") or is_escape(key):
return
index = max(0, min(index, max(0, len(shown) - 1)))
@@ -923,24 +766,24 @@ class Editor:
"Enter Aendern | q zurueck")
for row, (label, value, _kind, _key) in enumerate(fields):
attr = curses.color_pair(C_SEL) if row == index else 0
_fill_row(win, 4 + row, 2,
fill_row(win, 4 + row, 2,
truncate(label, 32).ljust(34) + truncate(str(value), max(4, width - 42)),
width, attr)
_put(win, 6 + len(fields), 2,
put(win, 6 + len(fields), 2,
"Platzhalter der Beschreibung: {group} {vmid} {name} {node} {type} "
"{datetime} {date} {time} {keep_time} {keep_count} {schedule}",
curses.color_pair(C_DIM))
win.refresh()
key = _read_key(win)
key = read_key(win)
if key in (curses.KEY_UP, "k"):
index -= 1
elif key in (curses.KEY_DOWN, "j"):
index += 1
elif _is_enter(key) or key == " ":
elif is_enter(key) or key == " ":
label, _value, kind, attribute = fields[index]
self._edit_global(win, label, kind, attribute)
elif key in ("q", "Q") or _is_escape(key):
elif key in ("q", "Q") or is_escape(key):
return
def _edit_global(self, win, label, kind, attribute):
@@ -994,7 +837,7 @@ class Editor:
height, width = self._frame(win, "Uebersicht: welche VM in welcher Gruppe?",
self._inventory_error or "%d Gaeste" % len(guests),
"Pfeiltasten Blaettern | r Neu laden | q zurueck")
_put(win, 3, 2, "VMID".ljust(8) + "Name".ljust(26) + "Status".ljust(10) + "Gruppen",
put(win, 3, 2, "VMID".ljust(8) + "Name".ljust(26) + "Status".ljust(10) + "Gruppen",
curses.A_BOLD)
rows = []
for guest in guests:
@@ -1009,7 +852,7 @@ class Editor:
break
guest, names = rows[position]
attr = 0 if names else curses.color_pair(C_DIM) | curses.A_DIM
_put(win, 4 + row, 2,
put(win, 4 + row, 2,
truncate(str(guest.vmid), 7).ljust(8)
+ truncate(guest.name or "-", 24).ljust(26)
+ truncate(guest.status, 9).ljust(10)
@@ -1017,7 +860,7 @@ class Editor:
attr)
win.refresh()
key = _read_key(win)
key = read_key(win)
if key == curses.KEY_DOWN:
top += 1
elif key == curses.KEY_UP:
@@ -1028,7 +871,7 @@ class Editor:
top -= visible
elif key == "r":
guests = self.guests(refresh=True)
elif key in ("q", "Q") or _is_escape(key):
elif key in ("q", "Q") or is_escape(key):
return
top = max(0, top)
@@ -1047,9 +890,6 @@ def _preview_next(group):
moment.strftime("%d.%m.%Y %H:%M"), describe(group))
def _fill_row(win, y, x, text, width, attr):
_put(win, y, x, str(text).ljust(max(0, width - x - 1))[: max(0, width - x - 1)], attr)
def run_editor(path):
editor = Editor(path)
+5
View File
@@ -0,0 +1,5 @@
"""Web-Oberflaeche von pvesnap: Snapshot-Inhalte im Browser herunterladen."""
from .server import main
__all__ = ["main"]
+6
View File
@@ -0,0 +1,6 @@
import sys
from .server import main
if __name__ == "__main__":
sys.exit(main())
+121
View File
@@ -0,0 +1,121 @@
"""HTML-Geruest und Stylesheet der Web-Oberflaeche.
Alles ist eingebettet - die Seite laedt nichts aus dem Internet nach und
funktioniert damit auch in abgeschotteten Netzen.
"""
from __future__ import annotations
STYLE = """
:root {
color-scheme: light dark;
--bg: #f6f7f9; --fg: #1c1f24; --muted: #6b7280; --line: #d8dce3;
--card: #ffffff; --accent: #2563eb; --accent-fg: #ffffff; --hover: #eef2ff;
}
@media (prefers-color-scheme: dark) {
:root {
--bg: #16181d; --fg: #e6e8ec; --muted: #9aa1ac; --line: #2c313a;
--card: #1e2127; --accent: #4b82f0; --accent-fg: #0b0d10; --hover: #232a39;
}
}
* { box-sizing: border-box; }
body {
margin: 0; background: var(--bg); color: var(--fg);
font: 15px/1.5 system-ui, -apple-system, "Segoe UI", Roboto, sans-serif;
}
header {
background: var(--card); border-bottom: 1px solid var(--line);
padding: .7rem 1rem; display: flex; gap: 1rem; align-items: center;
flex-wrap: wrap; position: sticky; top: 0; z-index: 5;
}
header h1 { font-size: 1rem; margin: 0; font-weight: 650; }
header .spacer { flex: 1 1 auto; }
main { padding: 1rem; max-width: 1200px; margin: 0 auto; }
a { color: var(--accent); text-decoration: none; }
a:hover { text-decoration: underline; }
.muted { color: var(--muted); }
.card {
background: var(--card); border: 1px solid var(--line);
border-radius: 10px; padding: .25rem .75rem; margin-bottom: 1rem;
}
table { width: 100%; border-collapse: collapse; }
th, td { text-align: left; padding: .45rem .5rem; border-bottom: 1px solid var(--line); }
th { font-size: .8rem; text-transform: uppercase; letter-spacing: .04em; color: var(--muted); }
tr:last-child td { border-bottom: 0; }
tbody tr:hover { background: var(--hover); }
td.size, th.size { text-align: right; white-space: nowrap; font-variant-numeric: tabular-nums; }
td.when, th.when { white-space: nowrap; color: var(--muted); font-size: .9em; }
td.pick { width: 2.2rem; }
.name { display: flex; gap: .5rem; align-items: center; word-break: break-all; }
.icon { width: 1.2rem; text-align: center; flex: 0 0 auto; }
button, .button {
font: inherit; padding: .4rem .8rem; border-radius: 7px; cursor: pointer;
border: 1px solid var(--line); background: var(--card); color: var(--fg);
}
button.primary, .button.primary { background: var(--accent); color: var(--accent-fg); border-color: transparent; }
button:hover, .button:hover { filter: brightness(1.08); text-decoration: none; }
.bar { display: flex; gap: .5rem; align-items: center; flex-wrap: wrap; margin: .75rem 0; }
.crumbs { display: flex; gap: .3rem; flex-wrap: wrap; align-items: center; word-break: break-all; }
.crumbs span { color: var(--muted); }
.note {
border-left: 3px solid var(--accent); background: var(--card);
padding: .6rem .8rem; border-radius: 0 8px 8px 0; margin-bottom: 1rem;
}
.warn { border-left-color: #d9534f; }
.pill {
display: inline-block; padding: .1rem .5rem; border-radius: 999px;
background: var(--hover); font-size: .82em; color: var(--muted);
}
input[type=checkbox] { width: 1.05rem; height: 1.05rem; }
.empty { padding: 2rem 1rem; text-align: center; color: var(--muted); }
@media (max-width: 640px) {
td.when, th.when { display: none; }
main { padding: .5rem; }
}
"""
# Auswahl-Kaestchen: "alle" umschalten und Anzahl anzeigen.
SCRIPT = """
function pvesnapInit() {
var all = document.getElementById('pick-all');
var boxes = Array.prototype.slice.call(document.querySelectorAll('.pick-one'));
var info = document.getElementById('pick-info');
function update() {
var n = boxes.filter(function (b) { return b.checked; }).length;
if (info) { info.textContent = n ? (n + ' ausgewaehlt') : ''; }
if (all) { all.checked = n > 0 && n === boxes.length; }
}
if (all) {
all.addEventListener('change', function () {
boxes.forEach(function (b) { b.checked = all.checked; });
update();
});
}
boxes.forEach(function (b) { b.addEventListener('change', update); });
update();
}
document.addEventListener('DOMContentLoaded', pvesnapInit);
"""
PAGE = """<!doctype html>
<html lang="de">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>%(title)s</title>
<style>%(style)s</style>
<script>%(script)s</script>
</head>
<body>
<header>
<h1><a href="%(home)s">pvesnap</a></h1>
<span class="muted">%(subtitle)s</span>
<span class="spacer"></span>
%(actions)s
</header>
<main>
%(body)s
</main>
</body>
</html>
"""
+775
View File
@@ -0,0 +1,775 @@
"""pvesnap-web - Snapshot-Inhalte im Browser durchsehen und herunterladen.
pvesnap-web Gast und Snapshot im Browser waehlen
pvesnap-web 802 <snapname> diesen Snapshot gleich oeffnen
pvesnap-web --path /verzeichnis beliebiges Verzeichnis (ohne Proxmox)
Beim Start wird eine Adresse mit Zugangsschluessel ausgegeben. Nur wer diese
Adresse kennt, kommt an die Daten - der Schluessel wandert danach in ein
Cookie. Es wird ausschliesslich gelesen; ueber die Oberflaeche laesst sich
nichts veraendern oder loeschen.
"""
from __future__ import annotations
import argparse
import html
import io
import logging
import os
import secrets
import socket
import stat
import sys
import threading
import time
import urllib.parse
import zipfile
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from .. import __version__
from ..proxmox import Proxmox, ProxmoxError, pvesh_available
from ..snapfs import (Session, SnapfsError, cleanup_leftovers, human_bytes,
list_snapshots)
from .assets import PAGE, SCRIPT, STYLE
log = logging.getLogger("pvesnap.web")
COOKIE = "pvesnap_token"
CHUNK = 256 * 1024
# ---------------------------------------------------------------------------
# Zustand
# ---------------------------------------------------------------------------
class AppState:
"""Was gerade geoeffnet ist. Es ist immer hoechstens ein Snapshot offen."""
def __init__(self, token, proxmox=None, fixed_path=""):
self.token = token
self.proxmox = proxmox
self.fixed_path = fixed_path
self.session = None
self.guest = None
self.snapname = ""
self.roots = [] # [(Beschriftung, Pfad)]
self.lock = threading.Lock()
self.notes = []
if fixed_path:
self.roots = [(fixed_path, os.path.realpath(fixed_path))]
@property
def open(self):
return bool(self.roots)
@property
def subtitle(self):
if self.fixed_path:
return html.escape(self.fixed_path)
if self.session:
return "%s &nbsp;·&nbsp; Snapshot <b>%s</b>" % (
html.escape(self.guest.label), html.escape(self.snapname))
return "Snapshot-Explorer %s" % __version__
def open_snapshot(self, guest, snapname):
with self.lock:
self.close_locked()
session = Session(self.proxmox, guest, snapname)
session.open()
session.prepare()
self.session = session
self.guest = guest
self.snapname = snapname
self.notes = list(session.notes)
self.roots = [(mountpoint, mountpoint)
for mountpoint in sorted(set(session.mounts.values()))]
def close_locked(self):
if self.session:
self.session.close()
self.session = None
self.guest = None
self.snapname = ""
self.notes = []
if not self.fixed_path:
self.roots = []
def close(self):
with self.lock:
self.close_locked()
def root(self, index):
if not self.roots:
return ""
index = max(0, min(index, len(self.roots) - 1))
return self.roots[index][1]
def resolve(self, index, relative):
"""Relativen Pfad in einen echten Pfad innerhalb des Snapshots umsetzen.
Alles ausserhalb der eingebundenen Wurzel wird abgewiesen - damit
laesst sich ueber die Adresszeile nichts anderes vom Host lesen.
"""
root = self.root(index)
if not root:
raise ValueError("Es ist kein Snapshot geoeffnet.")
candidate = os.path.realpath(os.path.join(root, (relative or "").lstrip("/")))
root_real = os.path.realpath(root)
if candidate != root_real and not candidate.startswith(root_real + os.sep):
raise ValueError("Pfad liegt ausserhalb des Snapshots.")
return candidate
def relative(self, index, path):
root = os.path.realpath(self.root(index))
path = os.path.realpath(path)
if path == root:
return ""
return os.path.relpath(path, root)
# ---------------------------------------------------------------------------
# Hilfsfunktionen fuer HTML
# ---------------------------------------------------------------------------
def esc(text):
return html.escape(str(text), quote=True)
def link(path_part, **params):
query = urllib.parse.urlencode({k: v for k, v in params.items() if v not in (None, "")})
return path_part + (("?" + query) if query else "")
def page(title, state, body, actions=""):
return PAGE % {
"title": esc(title),
"style": STYLE,
"script": SCRIPT,
"home": "/",
"subtitle": state.subtitle,
"actions": actions,
"body": body,
}
def icon_for(entry_is_dir, is_link, is_special):
if is_special:
return ""
if is_link:
return ""
return "📁" if entry_is_dir else "📄"
# ---------------------------------------------------------------------------
# HTTP
# ---------------------------------------------------------------------------
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
server_version = "pvesnap/" + __version__
state = None # wird beim Start gesetzt
# -- Grundlagen -------------------------------------------------------
def log_message(self, fmt, *args):
log.info("%s %s", self.address_string(), fmt % args)
def _send(self, body, status=200, content_type="text/html; charset=utf-8",
extra_headers=None):
if isinstance(body, str):
body = body.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Referrer-Policy", "no-referrer")
for key, value in (extra_headers or {}).items():
self.send_header(key, value)
self.end_headers()
if self.command != "HEAD":
self.wfile.write(body)
def _redirect(self, target, cookie=None):
self.send_response(303)
self.send_header("Location", target)
self.send_header("Content-Length", "0")
if cookie:
self.send_header("Set-Cookie",
"%s=%s; Path=/; HttpOnly; SameSite=Strict; Max-Age=86400"
% (COOKIE, cookie))
self.end_headers()
def _error(self, message, status=400):
body = page("Fehler", self.state,
'<div class="note warn"><b>Das ging nicht:</b><br>%s</div>'
'<p><a class="button" href="/">Zurueck zum Anfang</a></p>'
% esc(message))
self._send(body, status=status)
# -- Zugang -----------------------------------------------------------
def _authorized(self, params):
supplied = ""
if "token" in params:
supplied = params["token"][0]
else:
cookies = self.headers.get("Cookie") or ""
for part in cookies.split(";"):
name, _, value = part.strip().partition("=")
if name == COOKIE:
supplied = value
return secrets.compare_digest(supplied, self.state.token), supplied
# -- Anfragen ---------------------------------------------------------
def do_HEAD(self):
self.do_GET()
def do_GET(self):
parsed = urllib.parse.urlparse(self.path)
params = urllib.parse.parse_qs(parsed.query)
route = parsed.path
ok, supplied = self._authorized(params)
if not ok:
self._send(page("Zugang", self.state,
'<div class="note warn">Zugangsschluessel fehlt oder ist '
'falsch.<br>Bitte die Adresse benutzen, die beim Start von '
'<code>pvesnap-web</code> ausgegeben wurde.</div>'),
status=403)
return
# Schluessel aus der Adresse in ein Cookie umziehen
if "token" in params and route not in ("/download", "/zip"):
clean = {k: v[0] for k, v in params.items() if k != "token"}
self._redirect(link(route, **clean), cookie=supplied)
return
try:
if route == "/":
self._page_home()
elif route == "/snapshots":
self._page_snapshots(params)
elif route == "/files":
self._page_files(params)
elif route == "/download":
self._download(params)
else:
self._error("Unbekannte Seite: %s" % route, status=404)
except (ValueError, SnapfsError, ProxmoxError) as exc:
self._error(str(exc))
except BrokenPipeError:
pass # Browser hat den Download abgebrochen
def do_POST(self):
parsed = urllib.parse.urlparse(self.path)
ok, _ = self._authorized(urllib.parse.parse_qs(parsed.query))
if not ok:
self._send("Zugang verweigert", status=403, content_type="text/plain")
return
length = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(length).decode("utf-8", "replace") if length else ""
form = urllib.parse.parse_qs(raw)
try:
if parsed.path == "/open":
self._open_snapshot(form)
elif parsed.path == "/close":
self.state.close()
self._redirect("/")
elif parsed.path == "/zip":
self._zip(form)
else:
self._error("Unbekannte Aktion", status=404)
except (ValueError, SnapfsError, ProxmoxError) as exc:
self._error(str(exc))
except BrokenPipeError:
pass
# -- Seiten -----------------------------------------------------------
def _page_home(self):
state = self.state
if state.open:
self._redirect(link("/files", fs=0, p=""))
return
if not state.proxmox:
self._error("Kein Verzeichnis geoeffnet.")
return
guests = state.proxmox.inventory(refresh=True)
rows = []
for guest in guests:
rows.append(
"<tr><td class='name'><span class='icon'>%s</span>"
"<a href='%s'>%s</a></td>"
"<td>%s</td><td>%s</td><td>%s</td><td class='muted'>%s</td></tr>"
% ("📦" if guest.type == "lxc" else "🖥",
esc(link("/snapshots", vmid=guest.vmid)),
esc(guest.name or ("VM %d" % guest.vmid)),
guest.vmid, "LXC" if guest.type == "lxc" else "VM",
esc(guest.status), esc(guest.node)))
body = (
"<h2>Gast auswaehlen</h2>"
"<div class='card'><table><thead><tr><th>Name</th><th>VMID</th>"
"<th>Typ</th><th>Status</th><th>Node</th></tr></thead><tbody>%s</tbody>"
"</table></div>" % ("".join(rows) or
"<tr><td colspan='5' class='empty'>Keine Gaeste gefunden.</td></tr>"))
self._send(page("pvesnap - Gaeste", state, body))
def _page_snapshots(self, params):
state = self.state
vmid = int((params.get("vmid") or ["0"])[0])
guest = next((g for g in state.proxmox.inventory() if g.vmid == vmid), None)
if guest is None:
self._error("Gast %d nicht gefunden." % vmid, status=404)
return
rows = []
for snap in list_snapshots(state.proxmox, guest):
when = (datetime.fromtimestamp(snap.snaptime).strftime("%d.%m.%Y %H:%M")
if snap.snaptime else "?")
rows.append(
"<tr><td class='name'><span class='icon'>🕘</span>%s</td>"
"<td class='when'>%s</td><td class='muted'>%s</td>"
"<td><form method='post' action='/open'>"
"<input type='hidden' name='vmid' value='%d'>"
"<input type='hidden' name='snap' value='%s'>"
"<button class='primary' type='submit'>Oeffnen</button>"
"</form></td></tr>"
% (esc(snap.name), esc(when),
esc(snap.description.replace("\n", " ")[:120]),
guest.vmid, esc(snap.name)))
body = (
"<p><a href='/'>&larr; Gaeste</a></p>"
"<h2>Snapshots von %s</h2>"
"<div class='note'>Das Oeffnen bindet den Snapshot schreibgeschuetzt "
"ein - das dauert ein paar Sekunden.</div>"
"<div class='card'><table><thead><tr><th>Snapshot</th><th class='when'>"
"Erstellt</th><th>Beschreibung</th><th></th></tr></thead><tbody>%s</tbody>"
"</table></div>"
% (esc(guest.label),
"".join(rows) or "<tr><td colspan='4' class='empty'>"
"Dieser Gast hat keine Snapshots.</td></tr>"))
self._send(page("Snapshots %s" % guest.label, state, body))
def _open_snapshot(self, form):
state = self.state
vmid = int((form.get("vmid") or ["0"])[0])
snapname = (form.get("snap") or [""])[0]
guest = next((g for g in state.proxmox.inventory() if g.vmid == vmid), None)
if guest is None or not snapname:
self._error("Gast oder Snapshot fehlt.")
return
log.info("Oeffne Snapshot %s von %s", snapname, guest.label)
state.open_snapshot(guest, snapname)
self._redirect(link("/files", fs=0, p=""))
def _page_files(self, params):
state = self.state
if not state.open:
self._redirect("/")
return
fs = int((params.get("fs") or ["0"])[0])
relative = (params.get("p") or [""])[0]
path = state.resolve(fs, relative)
if not os.path.isdir(path):
self._redirect(link("/download", fs=fs, p=relative))
return
entries = []
try:
with os.scandir(path) as scan:
for item in scan:
entries.append(self._describe(item))
except OSError as exc:
self._error("%s: %s" % (path, exc.strerror or exc))
return
entries.sort(key=lambda e: (0 if e["dir"] else 1, e["name"].lower()))
body = [self._crumbs(fs, relative), self._filesystems(fs)]
for note in state.notes:
body.append("<div class='note'>%s</div>" % esc(note))
rows = []
if relative:
parent = os.path.dirname(relative.rstrip("/"))
rows.append("<tr><td class='pick'></td><td class='name'>"
"<span class='icon'>↩</span><a href='%s'>..</a></td>"
"<td class='size'></td><td class='when'></td></tr>"
% esc(link("/files", fs=fs, p=parent)))
for entry in entries:
child = os.path.join(relative, entry["name"]) if relative else entry["name"]
target = link("/files" if entry["dir"] else "/download", fs=fs, p=child)
rows.append(
"<tr><td class='pick'>%s</td>"
"<td class='name'><span class='icon'>%s</span><a href='%s'>%s</a></td>"
"<td class='size'>%s</td><td class='when'>%s</td></tr>"
% ("" if entry["special"] else
"<input class='pick-one' type='checkbox' name='p' value='%s' form='zipform'>"
% esc(child),
icon_for(entry["dir"], entry["link"], entry["special"]),
esc(target), esc(entry["name"]),
"&lt;DIR&gt;" if entry["dir"] else
("&lt;SPEZ&gt;" if entry["special"] else esc(human_bytes(entry["size"]))),
esc(entry["when"])))
body.append(
"<form id='zipform' method='post' action='/zip'>"
"<input type='hidden' name='fs' value='%d'>"
"<input type='hidden' name='base' value='%s'>"
"</form>" % (fs, esc(relative)))
body.append(
"<div class='bar'>"
"<label class='pill'><input type='checkbox' id='pick-all'> alle</label>"
"<span class='muted' id='pick-info'></span>"
"<button class='primary' type='submit' form='zipform'>"
"Auswahl als ZIP laden</button>"
"<span class='muted'>Einzelne Dateien: einfach anklicken.</span>"
"</div>")
body.append(
"<div class='card'><table><thead><tr><th class='pick'></th><th>Name</th>"
"<th class='size'>Groesse</th><th class='when'>Geaendert</th></tr></thead>"
"<tbody>%s</tbody></table></div>"
% ("".join(rows) or "<tr><td colspan='4' class='empty'>Leeres Verzeichnis</td></tr>"))
actions = ("<form method='post' action='/close' style='margin:0'>"
"<button type='submit'>Snapshot schliessen</button></form>"
if not state.fixed_path else "")
self._send(page("pvesnap - %s" % (relative or "/"), state,
"".join(body), actions=actions))
@staticmethod
def _describe(item):
try:
info = item.stat(follow_symlinks=False)
mode = info.st_mode
is_link = stat.S_ISLNK(mode)
if is_link:
try:
is_dir = stat.S_ISDIR(item.stat(follow_symlinks=True).st_mode)
except OSError:
is_dir = False
special = False
else:
is_dir = stat.S_ISDIR(mode)
special = not (is_dir or stat.S_ISREG(mode))
return {"name": item.name, "dir": is_dir, "link": is_link,
"special": special, "size": info.st_size,
"when": datetime.fromtimestamp(info.st_mtime).strftime("%d.%m.%y %H:%M")}
except OSError:
return {"name": item.name, "dir": False, "link": False, "special": True,
"size": 0, "when": ""}
def _crumbs(self, fs, relative):
parts = [p for p in (relative or "").split("/") if p]
crumbs = ["<a href='%s'>%s</a>" % (esc(link("/files", fs=fs, p="")), "")]
walked = ""
for part in parts:
walked = os.path.join(walked, part)
crumbs.append("<span>/</span><a href='%s'>%s</a>"
% (esc(link("/files", fs=fs, p=walked)), esc(part)))
return "<div class='crumbs'>%s</div>" % "".join(crumbs)
def _filesystems(self, current):
state = self.state
if len(state.roots) < 2:
return ""
items = []
for index, (label, _path) in enumerate(state.roots):
if index == current:
items.append("<span class='pill'>%s</span>" % esc(label))
else:
items.append("<a class='button' href='%s'>%s</a>"
% (esc(link("/files", fs=index, p="")), esc(label)))
return "<div class='bar'><span class='muted'>Dateisystem:</span>%s</div>" % "".join(items)
# -- Herunterladen ----------------------------------------------------
def _download(self, params):
state = self.state
fs = int((params.get("fs") or ["0"])[0])
relative = (params.get("p") or [""])[0]
path = state.resolve(fs, relative)
if os.path.isdir(path):
self._zip_response([relative], fs, os.path.basename(relative) or "snapshot")
return
try:
info = os.stat(path)
except OSError as exc:
self._error("%s: %s" % (relative, exc.strerror or exc), status=404)
return
if not stat.S_ISREG(info.st_mode):
self._error("Nur regulaere Dateien lassen sich herunterladen.")
return
name = os.path.basename(path) or "datei"
self.send_response(200)
self.send_header("Content-Type", "application/octet-stream")
self.send_header("Content-Length", str(info.st_size))
self.send_header("Content-Disposition",
"attachment; filename*=UTF-8''%s"
% urllib.parse.quote(name, safe=""))
self.send_header("X-Content-Type-Options", "nosniff")
self.end_headers()
if self.command == "HEAD":
return
with open(path, "rb") as handle:
while True:
chunk = handle.read(CHUNK)
if not chunk:
break
self.wfile.write(chunk)
def _zip(self, form):
fs = int((form.get("fs") or ["0"])[0])
base = (form.get("base") or [""])[0]
selected = form.get("p") or []
if not selected:
self._error("Nichts ausgewaehlt.")
return
name = os.path.basename(base) or "snapshot"
self._zip_response(selected, fs, name)
def _zip_response(self, relatives, fs, name):
"""ZIP direkt in die Verbindung schreiben - ohne Zwischendatei."""
state = self.state
paths = []
for relative in relatives:
path = state.resolve(fs, relative)
if os.path.lexists(path):
paths.append((relative, path))
if not paths:
self._error("Nichts gefunden.", status=404)
return
filename = "%s-%s.zip" % (name or "snapshot", time.strftime("%Y%m%d-%H%M%S"))
self.send_response(200)
self.send_header("Content-Type", "application/zip")
self.send_header("Transfer-Encoding", "chunked")
self.send_header("Content-Disposition",
"attachment; filename*=UTF-8''%s"
% urllib.parse.quote(filename, safe=""))
self.end_headers()
if self.command == "HEAD":
return
sink = _ChunkedSink(self.wfile)
skipped = 0
try:
with zipfile.ZipFile(sink, "w", compression=zipfile.ZIP_DEFLATED,
compresslevel=1, allowZip64=True) as archive:
for relative, path in paths:
skipped += self._add_to_zip(archive, path,
os.path.basename(relative.rstrip("/"))
or "inhalt")
sink.finish()
except BrokenPipeError:
log.info("Download abgebrochen: %s", filename)
if skipped:
log.info("%s: %d Spezialdatei(en) uebersprungen", filename, skipped)
def _add_to_zip(self, archive, path, arcname):
skipped = 0
try:
info = os.lstat(path)
except OSError:
return 1
if stat.S_ISLNK(info.st_mode):
entry = zipfile.ZipInfo(arcname)
entry.create_system = 3
entry.external_attr = (0xA000 | 0o777) << 16 # Symlink im ZIP
archive.writestr(entry, os.readlink(path))
return 0
if stat.S_ISDIR(info.st_mode):
empty = True
try:
for name in sorted(os.listdir(path)):
empty = False
skipped += self._add_to_zip(archive, os.path.join(path, name),
os.path.join(arcname, name))
except OSError:
skipped += 1
if empty:
archive.writestr(zipfile.ZipInfo(arcname.rstrip("/") + "/"), b"")
return skipped
if not stat.S_ISREG(info.st_mode):
return 1 # Geraetedateien, Sockets, FIFOs
try:
archive.write(path, arcname)
except OSError:
skipped += 1
return skipped
class _ChunkedSink:
"""Datei-aehnliches Objekt, das direkt in die HTTP-Verbindung schreibt.
zipfile kommt mit nicht positionierbaren Stroemen zurecht, braucht aber
tell(); deshalb wird die Position selbst mitgezaehlt.
"""
def __init__(self, wfile):
self.wfile = wfile
self.position = 0
self.closed = False
def write(self, data):
if not data:
return 0
self.wfile.write(b"%x\r\n" % len(data))
self.wfile.write(data)
self.wfile.write(b"\r\n")
self.position += len(data)
return len(data)
def tell(self):
return self.position
def flush(self):
try:
self.wfile.flush()
except (BrokenPipeError, ValueError):
pass
@staticmethod
def seekable():
return False
def finish(self):
if not self.closed:
self.closed = True
self.wfile.write(b"0\r\n\r\n")
self.flush()
# ---------------------------------------------------------------------------
# Start
# ---------------------------------------------------------------------------
def _addresses(port):
"""Erreichbare Adressen des Hosts - fuer die Ausgabe beim Start."""
found = []
try:
hostname = socket.gethostname()
for info in socket.getaddrinfo(hostname, port, proto=socket.IPPROTO_TCP):
address = info[4][0]
if address not in found and not address.startswith("127."):
found.append(address)
except OSError:
pass
return found or ["<IP-des-Hosts>"]
def build_parser():
parser = argparse.ArgumentParser(
prog="pvesnap-web",
description="Snapshot-Inhalte im Browser durchsehen und herunterladen.")
parser.add_argument("vmid", nargs="?", type=int, help="VMID des Gastes")
parser.add_argument("snapshot", nargs="?", help="Name des Snapshots")
parser.add_argument("-b", "--bind", default="0.0.0.0",
help="Adresse, auf der gelauscht wird (Vorgabe: %(default)s)")
parser.add_argument("-P", "--port", type=int, default=8823,
help="Port (Vorgabe: %(default)s)")
parser.add_argument("-p", "--path",
help="beliebiges Verzeichnis freigeben, ohne Proxmox zu fragen")
parser.add_argument("-t", "--token", help="Zugangsschluessel selbst vorgeben")
parser.add_argument("--no-token", action="store_true",
help="ohne Zugangsschluessel (nur in vertrauenswuerdigen Netzen!)")
parser.add_argument("--cleanup", action="store_true",
help="haengengebliebene Einbindungen aufraeumen und beenden")
parser.add_argument("-v", "--verbose", action="store_true")
parser.add_argument("-V", "--version", action="version",
version="pvesnap-web %s" % __version__)
return parser
def main(argv=None):
args = build_parser().parse_args(argv)
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%H:%M:%S")
if args.cleanup:
removed = cleanup_leftovers(verbose=True)
print("%d Eintrag/Eintraege aufgeraeumt." % len(removed))
return 0
proxmox = None
if not args.path:
if not pvesh_available():
print("'pvesh' nicht gefunden - pvesnap-web laeuft auf dem Proxmox-Host.\n"
"Zum Ausprobieren: pvesnap-web --path /ein/verzeichnis", file=sys.stderr)
return 3
if os.geteuid() != 0:
print("pvesnap-web braucht root-Rechte (Snapshots einbinden und mounten).",
file=sys.stderr)
return 4
proxmox = Proxmox()
elif not os.path.isdir(args.path):
print("Kein Verzeichnis: %s" % args.path, file=sys.stderr)
return 2
token = "" if args.no_token else (args.token or secrets.token_urlsafe(18))
state = AppState(token=token, proxmox=proxmox, fixed_path=args.path or "")
if args.vmid and args.snapshot and proxmox:
guest = next((g for g in proxmox.inventory() if g.vmid == args.vmid), None)
if guest is None:
print("Gast %d nicht gefunden." % args.vmid, file=sys.stderr)
return 2
log.info("Binde Snapshot %s von %s ein ...", args.snapshot, guest.label)
try:
state.open_snapshot(guest, args.snapshot)
except (SnapfsError, ProxmoxError) as exc:
print("Snapshot nicht zu oeffnen: %s" % exc, file=sys.stderr)
return 5
if args.no_token:
Handler._authorized = lambda self, params: (True, "")
Handler.state = state
server = ThreadingHTTPServer((args.bind, args.port), Handler)
server.daemon_threads = True
suffix = ("?token=%s" % token) if token else ""
lines = ["", "pvesnap-web laeuft. Im Browser oeffnen:", ""]
for address in _addresses(args.port):
lines.append(" http://%s:%d/%s" % (address, args.port, suffix))
if args.bind in ("127.0.0.1", "localhost", "::1"):
lines.append(" http://%s:%d/%s" % (args.bind, args.port, suffix))
lines += ["", " Beenden mit Strg-C."]
if not token:
lines.append(" ACHTUNG: ohne Zugangsschluessel gestartet.")
lines.append("")
# flush: sonst haengt die Adresse mit dem Schluessel im Puffer, sobald die
# Ausgabe nicht direkt im Terminal landet (Pipe, systemd, tee).
print("\n".join(lines), flush=True)
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nBeende ...")
finally:
server.server_close()
state.close()
return 0
if __name__ == "__main__": # pragma: no cover
sys.exit(main())
+5 -3
View File
@@ -61,10 +61,12 @@ fi
systemctl daemon-reload 2>/dev/null || true
systemctl reset-failed pvesnap.service 2>/dev/null || true
if [ -e "$BIN" ]; then
info "Entferne $BIN"
rm -f "$BIN"
for command in "$BIN" "${BIN}-explorer" "${BIN}-web"; do
if [ -e "$command" ]; then
info "Entferne $command"
rm -f "$command"
fi
done
if [ -d "$LIB_DIR" ]; then
info "Entferne $LIB_DIR"
rm -rf "$LIB_DIR"