v0.0.1 first commit
This commit is contained in:
@@ -0,0 +1,369 @@
|
||||
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,
|
||||
"stfv-backup-client",
|
||||
"stfv-backup-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,
|
||||
"stfv-backup-client",
|
||||
"stfv-backup-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():
|
||||
return True, "Lokaler Pfad erreichbar."
|
||||
return False, "Lokaler Pfad ist nicht erreichbar oder kein Verzeichnis."
|
||||
|
||||
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)
|
||||
ftp.cwd(target.subpath or "/")
|
||||
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)
|
||||
sftp.listdir(target.subpath or "/")
|
||||
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_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,
|
||||
"stfv-backup-client",
|
||||
"stfv-backup-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
|
||||
Reference in New Issue
Block a user