x3
226108f4dc
paramikos getcwd() liefert None, solange kein chdir() erfolgt ist. Der
Rueckfall auf "." landete als remote_path in der App, woraus _join()/_abs()
Pfade wie "/./datei" bauten. Der Server loeste die relativ zum Dateisystem-
Root statt zum Home auf -> "No such file" beim Download, "Permission denied"
beim Upload. Das Listing funktionierte, weil listdir(".") serverseitig
relativ zum Home aufgeloest wird -- daher fiel es erst beim Transfer auf.
- connect()/chdir()/listdir() loesen den Pfad ueber sftp.normalize() auf
- _abs() haengt relative Pfade an das CWD an, statt blind "/" voranzustellen
- _copy_tree(): die beiden mkdir-Zweige waren vertauscht. Beim Download
wurde backend.mkdir() auf einen lokalen Pfad angewendet (Verzeichnis
entstand auf dem Server), beim Upload os.makedirs() auf einen entfernten.
Ordner-Transfers waren dadurch in beide Richtungen defekt.
Verifiziert per End-to-End-Test gegen einen echten sshd/SFTP-Server:
Datei- und rekursiver Ordner-Transfer in beide Richtungen.
Closes #3
124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
"""SFTP-Transport basierend auf ``paramiko`` (SSH-FTP)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import stat
|
||
from pathlib import PurePosixPath
|
||
|
||
import paramiko
|
||
|
||
from .models import RemoteEntry
|
||
|
||
|
||
class SFTPBackend:
|
||
"""Adapter, der ``paramiko``/SFTP auf das Transport-Protokoll mappt."""
|
||
|
||
def __init__(self) -> None:
|
||
self._ssh: paramiko.SSHClient | None = None
|
||
self._sftp: paramiko.SFTPClient | None = None
|
||
self._cwd = "."
|
||
|
||
def connect(self, host: str, port: int, username: str, password: str) -> None:
|
||
self._ssh = paramiko.SSHClient()
|
||
self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
self._ssh.connect(
|
||
host,
|
||
port=port,
|
||
username=username,
|
||
password=password,
|
||
timeout=30,
|
||
)
|
||
self._sftp = self._ssh.open_sftp()
|
||
self._cwd = self._resolve(".")
|
||
|
||
def _resolve(self, path: str) -> str:
|
||
"""Löst einen Pfad serverseitig zu einem absoluten Pfad auf.
|
||
|
||
``paramiko.SFTPClient.getcwd()`` liefert ``None``, solange kein
|
||
explizites ``chdir()`` erfolgt ist – ein Rückfall auf ``"."`` würde
|
||
später zu Pfaden wie ``/./datei`` führen, die der Server relativ zum
|
||
Dateisystem-Root statt zum Home-Verzeichnis auflöst.
|
||
"""
|
||
assert self._sftp is not None
|
||
try:
|
||
return self._sftp.normalize(path)
|
||
except OSError:
|
||
return path if path.startswith("/") else "/"
|
||
|
||
def listdir(self, path: str) -> list[RemoteEntry]:
|
||
assert self._sftp is not None
|
||
entries: list[RemoteEntry] = []
|
||
for attr in self._sftp.listdir_attr(path):
|
||
mode = attr.st_mode
|
||
kind = "dir" if stat.S_ISDIR(mode) else "file"
|
||
entries.append(
|
||
RemoteEntry(
|
||
name=attr.filename,
|
||
kind=kind,
|
||
size=attr.st_size,
|
||
modified=attr.st_mtime,
|
||
)
|
||
)
|
||
self._cwd = self._resolve(path)
|
||
return entries
|
||
|
||
def chdir(self, path: str) -> None:
|
||
assert self._sftp is not None
|
||
self._sftp.chdir(path)
|
||
self._cwd = self._resolve(".")
|
||
|
||
def pwd(self) -> str:
|
||
return self._cwd
|
||
|
||
def _abs(self, path: str) -> str:
|
||
"""Baut einen absoluten Pfad – relative Pfade gelten zum aktuellen CWD."""
|
||
if not path:
|
||
return self._cwd or "/"
|
||
base = self._cwd if self._cwd.startswith("/") else "/"
|
||
return str(PurePosixPath(base) / path)
|
||
|
||
def download(self, remote: str, local: str) -> None:
|
||
assert self._sftp is not None
|
||
self._sftp.get(self._abs(remote), local)
|
||
|
||
def upload(self, local: str, remote: str) -> None:
|
||
assert self._sftp is not None
|
||
self._sftp.put(local, self._abs(remote))
|
||
|
||
def mkdir(self, path: str) -> None:
|
||
assert self._sftp is not None
|
||
try:
|
||
self._sftp.mkdir(self._abs(path))
|
||
except IOError:
|
||
pass
|
||
|
||
def remove(self, path: str) -> None:
|
||
assert self._sftp is not None
|
||
self._sftp.remove(self._abs(path))
|
||
|
||
def rmdir(self, path: str) -> None:
|
||
assert self._sftp is not None
|
||
self._recurse_rmdir(self._abs(path))
|
||
|
||
def _recurse_rmdir(self, path: str) -> None:
|
||
assert self._sftp is not None
|
||
for attr in self._sftp.listdir_attr(path):
|
||
child = f"{path.rstrip('/')}/{attr.filename}"
|
||
if stat.S_ISDIR(attr.st_mode):
|
||
self._recurse_rmdir(child)
|
||
else:
|
||
self._sftp.remove(child)
|
||
self._sftp.rmdir(path)
|
||
|
||
def rename(self, src: str, dst: str) -> None:
|
||
assert self._sftp is not None
|
||
self._sftp.rename(self._abs(src), self._abs(dst))
|
||
|
||
def close(self) -> None:
|
||
if self._sftp is not None:
|
||
self._sftp.close()
|
||
self._sftp = None
|
||
if self._ssh is not None:
|
||
self._ssh.close()
|
||
self._ssh = None
|