460 lines
16 KiB
Python
460 lines
16 KiB
Python
from dataclasses import dataclass
|
|
from ftplib import FTP
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
|
|
import paramiko
|
|
from smb.SMBConnection import SMBConnection
|
|
import smbclient
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TargetConnection:
|
|
protocol: str
|
|
host: str
|
|
port: int | None
|
|
share: str
|
|
subpath: str
|
|
username: str
|
|
password: str
|
|
|
|
|
|
def normalize_subpath(value: str) -> str:
|
|
path = value.strip()
|
|
if not path:
|
|
return "/"
|
|
if not path.startswith("/"):
|
|
return f"/{path}"
|
|
return path
|
|
|
|
|
|
def _smb_list_path(path: str) -> str:
|
|
normalized = normalize_subpath(path)
|
|
trimmed = normalized.strip("/")
|
|
return "/" if not trimmed else trimmed
|
|
|
|
|
|
def _build_child_path(parent: str, name: str) -> str:
|
|
p = normalize_subpath(parent).rstrip("/")
|
|
if not p:
|
|
p = "/"
|
|
if p == "/":
|
|
return f"/{name}"
|
|
return f"{p}/{name}"
|
|
|
|
|
|
def _sort_entries(items: list[dict[str, str]]) -> list[dict[str, str]]:
|
|
return sorted(items, key=lambda x: (0 if x["entry_type"] == "directory" else 1, x["name"].lower()))
|
|
|
|
|
|
def _smb_unc_path(target: TargetConnection, subpath: str) -> str:
|
|
share = target.share.strip().strip("\\/")
|
|
clean_subpath = normalize_subpath(subpath).strip("/")
|
|
if clean_subpath:
|
|
rel = clean_subpath.replace("/", "\\")
|
|
return f"\\\\{target.host}\\{share}\\{rel}"
|
|
return f"\\\\{target.host}\\{share}"
|
|
|
|
|
|
def _list_smb_shares_with_smbprotocol(target: TargetConnection) -> tuple[bool, str, list[str]]:
|
|
try:
|
|
smbclient.register_session(
|
|
server=target.host,
|
|
username=target.username,
|
|
password=target.password,
|
|
port=target.port or 445,
|
|
)
|
|
# Manche Server erlauben das Listing von \\host als Share-Quelle.
|
|
entries = smbclient.listdir(f"\\\\{target.host}\\")
|
|
shares = sorted({str(item).strip("\\/") for item in entries if str(item).strip("\\/")})
|
|
if shares:
|
|
return True, "SMB-Shares erfolgreich geladen.", shares
|
|
return False, "Keine Shares gefunden.", []
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"smbprotocol Fehler: {exc}", []
|
|
|
|
|
|
def _list_smb_shares_with_cli(target: TargetConnection) -> tuple[bool, str, list[str]]:
|
|
smbclient_bin = shutil.which("smbclient")
|
|
if smbclient_bin is None:
|
|
return False, "smbclient CLI ist nicht installiert.", []
|
|
|
|
cmd = [
|
|
smbclient_bin,
|
|
"-g",
|
|
"-L",
|
|
f"//{target.host}",
|
|
"-U",
|
|
f"{target.username}%{target.password}",
|
|
"-m",
|
|
"SMB3",
|
|
]
|
|
if target.port:
|
|
cmd.extend(["-p", str(target.port)])
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
|
|
if result.returncode != 0:
|
|
err = (result.stderr or result.stdout or "Unbekannter Fehler").strip()
|
|
return False, f"smbclient Fehler: {err}", []
|
|
|
|
shares: list[str] = []
|
|
for line in result.stdout.splitlines():
|
|
parts = [p.strip() for p in line.split("|")]
|
|
if len(parts) < 3:
|
|
continue
|
|
if parts[0].lower() != "disk":
|
|
continue
|
|
if parts[1]:
|
|
shares.append(parts[1])
|
|
|
|
uniq = sorted(set(shares))
|
|
if not uniq:
|
|
return False, "Es wurden keine Disk-Shares gefunden.", []
|
|
return True, "SMB-Shares erfolgreich geladen.", uniq
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"smbclient Aufruf fehlgeschlagen: {exc}", []
|
|
|
|
|
|
def list_smb_shares(target: TargetConnection) -> tuple[bool, str, list[str]]:
|
|
if target.protocol.upper() != "SMB":
|
|
return False, "Share-Liste ist nur fuer SMB verfuegbar.", []
|
|
if not target.host or not target.username or not target.password:
|
|
return False, "Fuer Share-Liste werden Host, Benutzer und Passwort benoetigt.", []
|
|
|
|
ok, msg, shares = _list_smb_shares_with_smbprotocol(target)
|
|
if ok:
|
|
return True, msg, shares
|
|
|
|
ok2, msg2, shares2 = _list_smb_shares_with_cli(target)
|
|
if ok2:
|
|
return True, msg2, shares2
|
|
|
|
return False, f"Share-Laden fehlgeschlagen. smbprotocol: {msg} | smbclient: {msg2}", []
|
|
|
|
|
|
def _test_smb_with_smbprotocol(target: TargetConnection) -> tuple[bool, str]:
|
|
try:
|
|
smbclient.register_session(
|
|
server=target.host,
|
|
username=target.username,
|
|
password=target.password,
|
|
port=target.port or 445,
|
|
)
|
|
unc = _smb_unc_path(target, target.subpath)
|
|
smbclient.listdir(unc)
|
|
return True, "SMB-Verbindung und Pfad sind erreichbar."
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"smbprotocol Fehler: {exc}"
|
|
|
|
|
|
def _test_smb_with_pysmb(target: TargetConnection) -> tuple[bool, str]:
|
|
conn = SMBConnection(
|
|
target.username,
|
|
target.password,
|
|
"backitup-client",
|
|
"backitup-server",
|
|
use_ntlm_v2=True,
|
|
is_direct_tcp=True,
|
|
)
|
|
try:
|
|
ok = conn.connect(target.host, target.port or 445, timeout=8)
|
|
if not ok:
|
|
return False, "pysmb: SMB-Verbindung konnte nicht aufgebaut werden."
|
|
conn.listPath(target.share, _smb_list_path(target.subpath))
|
|
return True, "SMB-Verbindung und Pfad sind erreichbar."
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"pysmb Fehler: {exc}"
|
|
finally:
|
|
try:
|
|
conn.close()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
def _browse_smb_with_smbprotocol(target: TargetConnection) -> tuple[bool, str, list[dict[str, str]]]:
|
|
current_path = normalize_subpath(target.subpath)
|
|
try:
|
|
smbclient.register_session(
|
|
server=target.host,
|
|
username=target.username,
|
|
password=target.password,
|
|
port=target.port or 445,
|
|
)
|
|
unc = _smb_unc_path(target, current_path)
|
|
entries_out: list[dict[str, str]] = []
|
|
|
|
if current_path != "/":
|
|
parent = "/" + "/".join(current_path.strip("/").split("/")[:-1])
|
|
parent = parent if parent else "/"
|
|
entries_out.append({"name": "..", "path": parent, "entry_type": "up"})
|
|
|
|
for item in smbclient.scandir(unc):
|
|
name = item.name
|
|
if name in {".", ".."}:
|
|
continue
|
|
entry_type = "directory" if item.is_dir() else "file"
|
|
entries_out.append({"name": name, "path": _build_child_path(current_path, name), "entry_type": entry_type})
|
|
|
|
ups = [x for x in entries_out if x["entry_type"] == "up"]
|
|
normal = [x for x in entries_out if x["entry_type"] != "up"]
|
|
return True, current_path, ups + _sort_entries(normal)
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"smbprotocol Fehler: {exc}", []
|
|
|
|
|
|
def _browse_smb_with_pysmb(target: TargetConnection) -> tuple[bool, str, list[dict[str, str]]]:
|
|
conn = SMBConnection(
|
|
target.username,
|
|
target.password,
|
|
"backitup-client",
|
|
"backitup-server",
|
|
use_ntlm_v2=True,
|
|
is_direct_tcp=True,
|
|
)
|
|
current_path = normalize_subpath(target.subpath)
|
|
|
|
try:
|
|
ok = conn.connect(target.host, target.port or 445, timeout=8)
|
|
if not ok:
|
|
return False, "pysmb: SMB-Verbindung konnte nicht aufgebaut werden.", []
|
|
|
|
entries = conn.listPath(target.share, _smb_list_path(current_path))
|
|
entries_out: list[dict[str, str]] = []
|
|
|
|
if current_path != "/":
|
|
parent = "/" + "/".join(current_path.strip("/").split("/")[:-1])
|
|
parent = parent if parent else "/"
|
|
entries_out.append({"name": "..", "path": parent, "entry_type": "up"})
|
|
|
|
for item in entries:
|
|
if item.filename in {".", ".."}:
|
|
continue
|
|
entry_type = "directory" if item.isDirectory else "file"
|
|
entries_out.append({"name": item.filename, "path": _build_child_path(current_path, item.filename), "entry_type": entry_type})
|
|
|
|
ups = [x for x in entries_out if x["entry_type"] == "up"]
|
|
normal = [x for x in entries_out if x["entry_type"] != "up"]
|
|
return True, current_path, ups + _sort_entries(normal)
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"pysmb Fehler: {exc}", []
|
|
finally:
|
|
try:
|
|
conn.close()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
def test_connection(target: TargetConnection) -> tuple[bool, str]:
|
|
protocol = target.protocol.upper()
|
|
|
|
if protocol == "LOCAL":
|
|
check_path = Path(target.subpath)
|
|
if check_path.exists() and (check_path.is_dir() or check_path.is_file()):
|
|
return True, "Lokaler Pfad erreichbar."
|
|
return False, "Lokaler Pfad nicht gefunden oder nicht zugänglich."
|
|
|
|
if protocol == "SMB":
|
|
if not target.share:
|
|
return False, "SMB-Share fehlt."
|
|
ok, msg = _test_smb_with_smbprotocol(target)
|
|
if ok:
|
|
return True, msg
|
|
ok2, msg2 = _test_smb_with_pysmb(target)
|
|
if ok2:
|
|
return True, msg2
|
|
return False, f"SMB-Test fehlgeschlagen. smbprotocol: {msg} | pysmb: {msg2}"
|
|
|
|
if protocol == "FTP":
|
|
ftp = FTP()
|
|
try:
|
|
ftp.connect(target.host, target.port or 21, timeout=8)
|
|
ftp.login(target.username, target.password)
|
|
subpath = target.subpath or "/"
|
|
try:
|
|
ftp.cwd(subpath)
|
|
except Exception: # noqa: BLE001
|
|
# subpath may be a file — try parent directory
|
|
parent = "/".join(subpath.rstrip("/").split("/")[:-1]) or "/"
|
|
ftp.cwd(parent)
|
|
return True, "FTP-Verbindung und Pfad sind erreichbar."
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"FTP-Test fehlgeschlagen: {exc}"
|
|
finally:
|
|
try:
|
|
ftp.quit()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
if protocol == "SFTP":
|
|
transport = None
|
|
sftp = None
|
|
try:
|
|
transport = paramiko.Transport((target.host, target.port or 22))
|
|
transport.connect(username=target.username, password=target.password)
|
|
sftp = paramiko.SFTPClient.from_transport(transport)
|
|
subpath = target.subpath or "/"
|
|
try:
|
|
sftp.listdir(subpath)
|
|
except Exception: # noqa: BLE001
|
|
# subpath may be a file — try parent directory
|
|
parent = "/".join(subpath.rstrip("/").split("/")[:-1]) or "/"
|
|
sftp.listdir(parent)
|
|
return True, "SFTP-Verbindung und Pfad sind erreichbar."
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"SFTP-Test fehlgeschlagen: {exc}"
|
|
finally:
|
|
if sftp is not None:
|
|
sftp.close()
|
|
if transport is not None:
|
|
transport.close()
|
|
|
|
return False, "Unbekanntes Protokoll."
|
|
|
|
|
|
def browse_ftp_directories(target: TargetConnection) -> tuple[bool, str, list[dict[str, str]]]:
|
|
ftp = FTP()
|
|
try:
|
|
ftp.connect(target.host, target.port or 21, timeout=8)
|
|
ftp.login(target.username, target.password)
|
|
current_path = normalize_subpath(target.subpath or "/")
|
|
try:
|
|
ftp.cwd(current_path)
|
|
except Exception:
|
|
current_path = "/"
|
|
ftp.cwd("/")
|
|
|
|
entries: list[dict[str, str]] = []
|
|
if current_path != "/":
|
|
parent = "/".join(current_path.rstrip("/").split("/")[:-1]) or "/"
|
|
entries.append({"name": "..", "path": parent, "entry_type": "up"})
|
|
|
|
raw_lines: list[str] = []
|
|
try:
|
|
ftp.retrlines("LIST", raw_lines.append)
|
|
except Exception:
|
|
pass
|
|
|
|
items: list[dict[str, str]] = []
|
|
for line in raw_lines:
|
|
parts = line.split(None, 8)
|
|
if len(parts) < 9:
|
|
continue
|
|
name = parts[8].strip()
|
|
if name in (".", ".."):
|
|
continue
|
|
is_dir = parts[0].startswith("d")
|
|
sub_path = _build_child_path(current_path, name)
|
|
items.append({"name": name, "path": sub_path, "entry_type": "directory" if is_dir else "file"})
|
|
|
|
entries.extend(_sort_entries(items))
|
|
return True, current_path, entries
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"FTP-Fehler: {exc}", []
|
|
finally:
|
|
try:
|
|
ftp.quit()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
|
|
def browse_sftp_directories(target: TargetConnection) -> tuple[bool, str, list[dict[str, str]]]:
|
|
import stat as stat_module
|
|
transport = None
|
|
sftp_client = None
|
|
try:
|
|
transport = paramiko.Transport((target.host, target.port or 22))
|
|
transport.connect(username=target.username, password=target.password)
|
|
sftp_client = paramiko.SFTPClient.from_transport(transport)
|
|
current_path = normalize_subpath(target.subpath or "/")
|
|
|
|
entries: list[dict[str, str]] = []
|
|
if current_path != "/":
|
|
parent = "/".join(current_path.rstrip("/").split("/")[:-1]) or "/"
|
|
entries.append({"name": "..", "path": parent, "entry_type": "up"})
|
|
|
|
items: list[dict[str, str]] = []
|
|
for attr in sftp_client.listdir_attr(current_path):
|
|
is_dir = stat_module.S_ISDIR(attr.st_mode or 0)
|
|
sub_path = _build_child_path(current_path, attr.filename)
|
|
items.append({"name": attr.filename, "path": sub_path, "entry_type": "directory" if is_dir else "file"})
|
|
|
|
entries.extend(_sort_entries(items))
|
|
return True, current_path, entries
|
|
except Exception as exc: # noqa: BLE001
|
|
return False, f"SFTP-Fehler: {exc}", []
|
|
finally:
|
|
if sftp_client is not None:
|
|
sftp_client.close()
|
|
if transport is not None:
|
|
transport.close()
|
|
|
|
|
|
def browse_smb_directories(target: TargetConnection) -> tuple[bool, str, list[dict[str, str]]]:
|
|
if target.protocol.upper() != "SMB":
|
|
return False, "SMB-Browser ist nur fuer SMB verfuegbar.", []
|
|
if not target.share:
|
|
return False, "SMB-Share fehlt.", []
|
|
|
|
ok, path_or_error, folders = _browse_smb_with_smbprotocol(target)
|
|
if ok:
|
|
return True, path_or_error, folders
|
|
|
|
ok2, path_or_error2, folders2 = _browse_smb_with_pysmb(target)
|
|
if ok2:
|
|
return True, path_or_error2, folders2
|
|
|
|
return (
|
|
False,
|
|
"SMB-Browsing fehlgeschlagen. "
|
|
f"smbprotocol: {path_or_error} | pysmb: {path_or_error2}",
|
|
[],
|
|
)
|
|
|
|
|
|
def create_smb_directory(target: TargetConnection, base_path: str, folder_name: str) -> tuple[bool, str]:
|
|
if target.protocol.upper() != "SMB":
|
|
return False, "Neuer Ordner kann nur bei SMB erstellt werden."
|
|
clean_name = folder_name.strip().strip("/").strip("\\")
|
|
if not clean_name:
|
|
return False, "Ordnername darf nicht leer sein."
|
|
if "/" in clean_name or "\\" in clean_name:
|
|
return False, "Ordnername darf keine Pfadtrenner enthalten."
|
|
|
|
parent = normalize_subpath(base_path)
|
|
full_path = _build_child_path(parent, clean_name)
|
|
|
|
try:
|
|
smbclient.register_session(
|
|
server=target.host,
|
|
username=target.username,
|
|
password=target.password,
|
|
port=target.port or 445,
|
|
)
|
|
smbclient.mkdir(_smb_unc_path(target, full_path))
|
|
return True, full_path
|
|
except Exception as exc: # noqa: BLE001
|
|
# Fallback via pysmb
|
|
conn = SMBConnection(
|
|
target.username,
|
|
target.password,
|
|
"backitup-client",
|
|
"backitup-server",
|
|
use_ntlm_v2=True,
|
|
is_direct_tcp=True,
|
|
)
|
|
try:
|
|
ok = conn.connect(target.host, target.port or 445, timeout=8)
|
|
if not ok:
|
|
return False, f"SMB-Verbindung fehlgeschlagen: {exc}"
|
|
conn.createDirectory(target.share, _smb_list_path(full_path))
|
|
return True, full_path
|
|
except Exception as exc2: # noqa: BLE001
|
|
return False, f"Ordner konnte nicht erstellt werden. smbprotocol: {exc} | pysmb: {exc2}"
|
|
finally:
|
|
try:
|
|
conn.close()
|
|
except Exception: # noqa: BLE001
|
|
pass
|