diff --git a/src/netmover/admin.py b/src/netmover/admin.py
index c58d784..c4e00a4 100644
--- a/src/netmover/admin.py
+++ b/src/netmover/admin.py
@@ -496,8 +496,6 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
     src = s.get(Mount, d.src_mount_id) if d.src_mount_id else None
     if src is None:
         errors["src_mount_id"] = "Quelle wählen."
-    elif not src.type.is_kernel_mount:
-        errors["src_mount_id"] = "SFTP-Anbindungen sind nur als Ziel möglich."
     try:
         d.src_subpath = safe_relpath("src_subpath", d.src_subpath)
     except ValidationError as exc:
@@ -530,6 +528,10 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
             errors["pattern"] = f"Ungültiger regulärer Ausdruck: {exc}"
     if d.stability_seconds >= d.op_timeout_seconds:
         errors["stability_seconds"] = "Muss kleiner als das Zeitlimit sein."
+    elif src is not None and src.type is MountType.SFTP and d.stability_seconds < 2:
+        errors["stability_seconds"] = (
+            "Bei SFTP-Quellen mindestens 2 s – der Server meldet Zeiten nur sekundengenau."
+        )
     if d.set_owner and not _OWNER_RE.fullmatch(d.set_owner):
         errors["set_owner"] = "Als „uid:gid“ angeben, z. B. „1000:1000“."
     try:
diff --git a/src/netmover/mover/destinations.py b/src/netmover/mover/destinations.py
index 3ef969c..575efcb 100644
--- a/src/netmover/mover/destinations.py
+++ b/src/netmover/mover/destinations.py
@@ -30,12 +30,13 @@ from abc import ABC, abstractmethod
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Literal
+from typing import Any, Literal, Protocol
 
 from netmover.models import CollisionPolicy
 from netmover.mounting.sftp import SftpSession
 from netmover.mover.fsops import rename_noreplace
 from netmover.mover.naming import collision_candidates
+from netmover.mover.sources import network_errors
 from netmover.timeouts import CancelToken
 
 CHUNK = 1024 * 1024
@@ -81,6 +82,13 @@ class CommitResult:
 Verify = Callable[[], None]
 
 
+class Readable(Protocol):
+    """Quelle einer Übertragung: ein lokaler ``Path`` oder eine Datei einer
+    :class:`~netmover.mover.sources.Source` (über einen kleinen Adapter)."""
+
+    def open(self, mode: str = ...) -> Any: ...
+
+
 class Destination(ABC):
     kind: FsKind
 
@@ -107,7 +115,7 @@ class Destination(ABC):
     def _ensure_parent(self, rel: str) -> None: ...
 
     @abstractmethod
-    def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]: ...
+    def _upload(self, src: Readable, tmp: str, size: int, token: CancelToken) -> list[str]: ...
 
     @abstractmethod
     def _apply_meta(self, rel: str, meta: MetaOptions) -> list[str]: ...
@@ -140,7 +148,7 @@ class Destination(ABC):
 
     def commit(
         self,
-        src: Path,
+        src: Readable,
         size: int,
         rel: str,
         policy: CollisionPolicy,
@@ -234,7 +242,7 @@ class LocalDestination(Destination):
     def _ensure_parent(self, rel: str) -> None:
         self._p(rel).parent.mkdir(mode=0o755, parents=True, exist_ok=True)
 
-    def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]:
+    def _upload(self, src: Readable, tmp: str, size: int, token: CancelToken) -> list[str]:
         written = 0
         fd = os.open(self._p(tmp), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
         try:
@@ -429,10 +437,10 @@ class SftpDestination(Destination):
             except FileNotFoundError:
                 self.client.mkdir(current, mode=0o755)
 
-    def _upload(self, src: Path, tmp: str, size: int, token: CancelToken) -> list[str]:
+    def _upload(self, src: Readable, tmp: str, size: int, token: CancelToken) -> list[str]:
         notes: list[str] = []
         written = 0
-        with self.client.open(self._p(tmp), "wb") as fh, src.open("rb") as fin:
+        with network_errors(), self.client.open(self._p(tmp), "wb") as fh, src.open("rb") as fin:
             fh.set_pipelined(True)
             while chunk := fin.read(CHUNK // 4):
                 token.check()
diff --git a/src/netmover/mover/engine.py b/src/netmover/mover/engine.py
index 1b45ced..37b995a 100644
--- a/src/netmover/mover/engine.py
+++ b/src/netmover/mover/engine.py
@@ -15,12 +15,12 @@ from __future__ import annotations
 
 import dataclasses
 import errno
-import os
-import stat
+import posixpath
 import time
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from pathlib import Path
+from typing import Any
 
 from netmover.logging_setup import event, get_logger
 from netmover.models import (
@@ -40,7 +40,8 @@ from netmover.mover.destinations import (
 )
 from netmover.mover.ledger import FailureInfo, Ledger
 from netmover.mover.naming import collision_candidates
-from netmover.mover.scan import BUSY, Candidate, FileState, Matcher, current_state, scan
+from netmover.mover.scan import BUSY, Candidate, FileState, Matcher
+from netmover.mover.sources import LocalSource, Source, SourceMissing
 from netmover.timeouts import Cancelled, CancelToken
 
 log = get_logger(__name__)
@@ -114,6 +115,17 @@ def _mb(n: int) -> str:
     return f"{n / 1024 / 1024:.1f} MB"
 
 
+@dataclass(frozen=True)
+class _SourceFile:
+    """Adapter: eine Datei einer :class:`Source` mit ``open()`` wie ein ``Path``."""
+
+    source: Source
+    rel: str
+
+    def open(self, mode: str = "rb") -> Any:
+        return self.source.open(self.rel)
+
+
 class Engine:
     def __init__(
         self,
@@ -140,20 +152,27 @@ class Engine:
     def execute(
         self,
         rule: RuleSnapshot,
-        src_root: Path,
+        source: Source | Path,
         dest: Destination,
         token: CancelToken,
         *,
         dry_run: bool = False,
+        quarantine_rel: str | None = None,
         quarantine_root: Path | None = None,
     ) -> RunResult:
+        """``source``: eine :class:`Source` oder (Kurzform) ein lokales Verzeichnis.
+        Quarantäne relativ zur Wurzel der Anbindung (``quarantine_rel``) oder – bei
+        der Kurzform – als Pfad innerhalb des Verzeichnisses (``quarantine_root``)."""
         result = RunResult(dry_run=dry_run)
+        src = source if isinstance(source, Source) else LocalSource(source)
+        if quarantine_root is not None and isinstance(src, LocalSource):
+            quarantine_rel = str(quarantine_root.relative_to(src.root))
         try:
-            self._execute(rule, src_root, dest, token, result, dry_run, quarantine_root)
+            self._execute(rule, src, dest, token, result, dry_run, quarantine_rel)
         except Cancelled:
             result.status = RunStatus.ERROR
             result.message = "Lauf abgebrochen"
-        except RunAborted as exc:
+        except (RunAborted, SourceMissing) as exc:
             result.status = RunStatus.ERROR
             result.message = str(exc)
         except TransferError as exc:
@@ -170,26 +189,23 @@ class Engine:
     def _execute(
         self,
         rule: RuleSnapshot,
-        src_root: Path,
+        src: Source,
         dest: Destination,
         token: CancelToken,
         result: RunResult,
         dry_run: bool,
-        quarantine_root: Path | None,
+        quarantine_rel: str | None,
     ) -> None:
-        if not src_root.is_dir():
-            raise RunAborted(f"Quellordner {src_root} existiert nicht")
+        src.check_root()
         if not dry_run:
             dest.prepare()
         token.check()
 
-        exclude: set[Path] = set()
-        if quarantine_root is not None:
-            exclude.add(quarantine_root.resolve())
-        if isinstance(dest, LocalDestination):
-            exclude.add(dest.root.resolve())  # Ziel innerhalb der Quelle nie erneut scannen
+        exclude = frozenset({quarantine_rel}) if quarantine_rel else frozenset()
+        if isinstance(dest, LocalDestination) and isinstance(src, LocalSource):
+            src.exclude_paths.add(dest.root.resolve())  # Ziel innerhalb der Quelle auslassen
         matcher = Matcher(rule.pattern, rule.pattern_type)
-        candidates = list(scan(src_root, matcher, rule.recursive, frozenset(exclude)))
+        candidates = src.scan(matcher, rule.recursive, exclude)
         result.seen = len(candidates)
         token.check()
 
@@ -227,13 +243,13 @@ class Engine:
 
         # Stabilität (Fallstrick 1): zwei frische Messungen (öffnen + fstat, nicht
         # der Verzeichnis-Cache) im Abstand des Stabilitätsfensters.
-        first: dict[str, FileState | None] = {c.relpath: current_state(c.path) for c in pending}
+        first: dict[str, FileState | None] = {c.relpath: src.state(c.relpath) for c in pending}
         if pending and rule.stability_seconds > 0:
             self._sleep(rule.stability_seconds)
             token.check()
         stable: list[Candidate] = []
         for c in pending:
-            before, st = first[c.relpath], current_state(c.path)
+            before, st = first[c.relpath], src.state(c.relpath)
             if before is None or st is None:
                 continue  # inzwischen verschwunden – nichts zu tun
             if st != before or st == BUSY:
@@ -255,7 +271,7 @@ class Engine:
         failures = self.ledger.failures(rule.id)
         for c in stable:
             token.check()
-            result.add(self._process(rule, c, dest, token, dry_run, quarantine_root, failures))
+            result.add(self._process(rule, c, src, dest, token, dry_run, quarantine_rel, failures))
 
         if unchanged:
             result.message = f"{unchanged} bereits übertragen"
@@ -263,9 +279,9 @@ class Engine:
     # -- einzelne Datei ------------------------------------------------------------------
 
     def _meta(
-        self, rule: RuleSnapshot, path: Path, dest: Destination
+        self, rule: RuleSnapshot, src: Source, rel: str, dest: Destination
     ) -> tuple[MetaOptions, list[str]]:
-        st = os.stat(path)
+        st = src.stat(rel)
         notes: list[str] = []
         owner = rule.set_owner
         if owner is not None:
@@ -280,17 +296,18 @@ class Engine:
                 if self._warn_once(f"{rule.id}:nfs-chown"):
                     reason = "Root-Squash aktiv" if rule.dst_root_squash else "Schreibtest fehlt"
                     notes.append(f"NFS-Ziel: chown übersprungen ({reason})")
-        mode = rule.set_mode if rule.set_mode is not None else stat.S_IMODE(st.st_mode)
-        return MetaOptions(st.st_atime_ns, st.st_mtime_ns, mode, owner), notes
+        mode = rule.set_mode if rule.set_mode is not None else st.mode
+        return MetaOptions(st.atime_ns, st.mtime_ns, mode, owner), notes
 
     def _process(
         self,
         rule: RuleSnapshot,
         c: Candidate,
+        src: Source,
         dest: Destination,
         token: CancelToken,
         dry_run: bool,
-        quarantine_root: Path | None,
+        quarantine_rel: str | None,
         failures: dict[str, FailureInfo],
     ) -> ItemResult:
         size = c.state.size
@@ -298,7 +315,7 @@ class Engine:
 
         # Quarantäne für wiederholt scheiternde Dateien
         if (
-            quarantine_root is not None
+            quarantine_rel is not None
             and failure is not None
             and failure.attempts >= rule.quarantine_after
             and (failure.size, failure.mtime_ns) == (c.state.size, c.state.mtime_ns)
@@ -307,11 +324,11 @@ class Engine:
                 return ItemResult(
                     c.relpath,
                     ItemStatus.PLANNED,
-                    str(quarantine_root / c.relpath),
+                    src.describe_root_rel(posixpath.join(quarantine_rel, c.relpath)),
                     size,
                     "würde in Quarantäne verschoben",
                 )
-            return self._quarantine(rule, c, quarantine_root, failure.attempts)
+            return self._quarantine(rule, c, src, quarantine_rel, failure.attempts)
 
         try:
             planned = dest.plan(c.relpath, rule.on_collision)
@@ -351,16 +368,25 @@ class Engine:
             )
 
         def verify() -> None:
-            if current_state(c.path) != c.state:
+            if src.state(c.relpath) != c.state:
                 raise SourceChanged("Quelle hat sich während der Übertragung geändert")
 
         try:
-            meta, notes = self._meta(rule, c.path, dest)
+            meta, notes = self._meta(rule, src, c.relpath, dest)
             res = None
-            if rule.action is RuleAction.MOVE:
-                res = dest.move_same_device(c.path, c.relpath, rule.on_collision, meta)
+            local = src.local_path(c.relpath)
+            if rule.action is RuleAction.MOVE and local is not None:
+                res = dest.move_same_device(local, c.relpath, rule.on_collision, meta)
             if res is None:
-                res = dest.commit(c.path, size, c.relpath, rule.on_collision, meta, verify, token)
+                res = dest.commit(
+                    _SourceFile(src, c.relpath),
+                    size,
+                    c.relpath,
+                    rule.on_collision,
+                    meta,
+                    verify,
+                    token,
+                )
         except CollisionSkip:
             return ItemResult(
                 c.relpath,
@@ -385,7 +411,7 @@ class Engine:
 
         if not res.source_consumed:
             try:
-                os.unlink(c.path)
+                src.delete(c.relpath)
             except OSError as exc:
                 # Ziel ist vollständig – Buchführung verhindert Duplikate im nächsten Lauf.
                 self.ledger.record_copy(rule.id, c.relpath, size, c.state.mtime_ns, target)
@@ -412,15 +438,18 @@ class Engine:
         return ItemResult(c.relpath, ItemStatus.FAILED, None, c.state.size, msg)
 
     def _quarantine(
-        self, rule: RuleSnapshot, c: Candidate, root: Path, attempts: int
+        self, rule: RuleSnapshot, c: Candidate, src: Source, q_rel: str, attempts: int
     ) -> ItemResult:
+        target = ""
         try:
             for cand in collision_candidates(c.relpath):
-                target = root / cand
-                if os.path.lexists(target):
+                target = posixpath.join(q_rel, cand)
+                if src.exists_root_rel(target):
+                    continue
+                try:
+                    src.move_to(c.relpath, target)
+                except FileExistsError:
                     continue
-                target.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
-                os.rename(c.path, target)
                 break
             else:
                 raise TransferError("kein freier Name im Quarantäneordner")
@@ -434,7 +463,7 @@ class Engine:
         return ItemResult(
             c.relpath,
             ItemStatus.QUARANTINED,
-            str(target),
+            src.describe_root_rel(target),
             c.state.size,
             f"nach {attempts} Fehlversuchen in Quarantäne verschoben",
         )
diff --git a/src/netmover/mover/runner.py b/src/netmover/mover/runner.py
index e36d6c4..e5ea375 100644
--- a/src/netmover/mover/runner.py
+++ b/src/netmover/mover/runner.py
@@ -34,6 +34,7 @@ from netmover.mover.destinations import (
 )
 from netmover.mover.engine import Engine, RuleSnapshot, RunResult
 from netmover.mover.ledger import Ledger
+from netmover.mover.sources import LocalSource, SftpSource, Source
 from netmover.timeouts import CancelToken, OperationTimeout, run_with_watchdog
 from netmover.validation import ValidationError, safe_abspath, safe_relpath
 
@@ -46,6 +47,7 @@ class RulePlan:
     name: str
     src_mount_id: int
     src_type: MountType
+    src_name: str
     src_subpath: str
     dst_mount_id: int | None
     dst_type: MountType | None
@@ -96,6 +98,7 @@ def load_plan(session: Session, rule_id: int) -> RulePlan:
         name=r.name,
         src_mount_id=r.src_mount_id,
         src_type=r.src_mount.type,
+        src_name=r.src_mount.name,
         src_subpath=safe_relpath("src_subpath", r.src_subpath),
         dst_mount_id=r.dst_mount_id,
         dst_type=dst.type if dst is not None else None,
@@ -166,10 +169,6 @@ class RuleRunner:
         rule_id: int,
         handed_over: list[bool],
     ) -> RunResult:
-        if not plan.src_type.is_kernel_mount:
-            return RunResult(
-                RunStatus.ERROR, dry, message="SFTP-Anbindungen sind nur als Ziel möglich"
-            )
         try:
             src_mp = self.mounts.check_alive(plan.src_mount_id)
             dst_mp = (
@@ -179,29 +178,33 @@ class RuleRunner:
             )
         except MountUnavailable as exc:
             return RunResult(RunStatus.ERROR, dry, message=str(exc), unavailable=True)
-        if src_mp is None:  # nur bei SFTP, oben bereits ausgeschlossen
-            return RunResult(RunStatus.ERROR, dry, message="Quelle ist kein Kernel-Mount")
-        src_root = src_mp / plan.src_subpath if plan.src_subpath else src_mp
-        quarantine_root = src_mp / plan.quarantine_path if plan.quarantine_path else None
 
         try:
+            make_source = self._source_factory(plan, src_mp)
+            src_root = src_mp / plan.src_subpath if src_mp and plan.src_subpath else src_mp
             make_dest = self._dest_factory(plan, dst_mp, src_root)
         except (ValidationError, TransferError, SftpError) as exc:
             return RunResult(RunStatus.ERROR, dry, message=str(exc))
 
         def work(token: CancelToken) -> RunResult:
-            dest = make_dest()
+            source = make_source()
+            try:
+                dest = make_dest()
+            except BaseException:
+                source.close()
+                raise
             try:
                 return self.engine.execute(
                     plan.snapshot,
-                    src_root,
+                    source,
                     dest,
                     token,
                     dry_run=dry,
-                    quarantine_root=quarantine_root,
+                    quarantine_rel=plan.quarantine_path,
                 )
             finally:
                 dest.close()
+                source.close()
 
         def finish() -> None:
             self._started.pop(rule_id, None)
@@ -230,8 +233,26 @@ class RuleRunner:
         paths = {path.absolute(), path.resolve()}
         return any(p == b or b in p.parents for p in paths for b in bases)
 
+    def _source_factory(self, plan: RulePlan, src_mp: Path | None) -> Callable[[], Source]:
+        if plan.src_type is MountType.SFTP:
+            scfg = self.mounts.sftp_config_for(plan.src_mount_id)
+            label = plan.src_name or "sftp"
+
+            def make_sftp() -> Source:
+                return SftpSource(connect(scfg), scfg.base_path, plan.src_subpath, label=label)
+
+            return make_sftp
+        if src_mp is None:
+            raise TransferError("Quell-Anbindung ist nicht eingehängt")
+        root = src_mp
+
+        def make_local() -> Source:
+            return LocalSource(root, plan.src_subpath)
+
+        return make_local
+
     def _dest_factory(
-        self, plan: RulePlan, dst_mp: Path | None, src_root: Path
+        self, plan: RulePlan, dst_mp: Path | None, src_root: Path | None
     ) -> Callable[[], Destination]:
         if plan.dst_mount_id is None:
             root = Path(safe_abspath("dst_path", plan.dst_path))
@@ -263,7 +284,7 @@ class RuleRunner:
             kind = "cifs" if plan.dst_type is MountType.CIFS else "nfs"
             create = True
 
-        if local_root.resolve() == src_root.resolve():
+        if src_root is not None and local_root.resolve() == src_root.resolve():
             raise ValidationError("dst_path", "Quelle und Ziel sind derselbe Ordner")
 
         def make_local() -> Destination:
diff --git a/src/netmover/mover/scan.py b/src/netmover/mover/scan.py
index eb9f882..5725dfb 100644
--- a/src/netmover/mover/scan.py
+++ b/src/netmover/mover/scan.py
@@ -34,7 +34,7 @@ class FileState:
 @dataclass(frozen=True)
 class Candidate:
     relpath: str  # POSIX, relativ zum Quellordner
-    path: Path
+    path: Path | None  # nur bei lokalen Quellen
     state: FileState
 
 
diff --git a/src/netmover/mover/sources.py b/src/netmover/mover/sources.py
new file mode 100644
index 0000000..5ae1f52
--- /dev/null
+++ b/src/netmover/mover/sources.py
@@ -0,0 +1,311 @@
+"""Quellen: eingehängtes Verzeichnis (SMB/NFS) oder SFTP.
+
+Die Engine greift nur über diese Schnittstelle auf die Quelle zu – scannen,
+frischen Zustand messen, lesen, löschen, in die Quarantäne verschieben. So gelten
+für SFTP-Quellen dieselben Garantien wie für eingehängte Freigaben.
+
+Adressierung: ``root`` ist die Wurzel der Anbindung, ``subdir`` der Quell-
+Unterordner der Regel. Kandidaten tragen Pfade relativ zu ``subdir`` (so landen sie
+im Ziel und in der Buchführung); die Quarantäne liegt relativ zu ``root``.
+
+SFTP-Besonderheit: Änderungszeiten kommen nur sekundengenau. Die Stabilitäts-
+prüfung stützt sich dort vor allem auf die Größe – mit einem Fenster ab 2 s
+unproblematisch.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import errno
+import os
+import posixpath
+import socket
+import stat
+from abc import ABC, abstractmethod
+from collections.abc import Iterator
+from dataclasses import dataclass
+from pathlib import Path
+from typing import IO, Any
+
+import paramiko
+
+from netmover.mounting.sftp import SftpSession
+from netmover.mover.fsops import rename_noreplace
+from netmover.mover.scan import (
+    IGNORED_DIRS,
+    Candidate,
+    FileState,
+    Matcher,
+    current_state,
+    is_ignored_name,
+    scan,
+)
+
+_NET_ERRORS = (paramiko.SSHException, EOFError, socket.timeout, ConnectionError)
+
+
+class SourceMissing(Exception):
+    """Quellordner existiert nicht (Lauf abbrechen, nichts anfassen)."""
+
+
+@dataclass(frozen=True)
+class SourceStat:
+    size: int
+    mtime_ns: int
+    atime_ns: int
+    mode: int  # Rechte-Bits
+
+
+def _join(*parts: str) -> str:
+    return posixpath.join(*[p for p in parts if p]) if any(parts) else ""
+
+
+@contextlib.contextmanager
+def network_errors() -> Iterator[None]:
+    """Verbindungsabbrüche bei SFTP als ``ENOTCONN`` melden – die Engine bricht den
+    Lauf dann ab (Anbindung gestört), statt jede Datei einzeln scheitern zu lassen."""
+    try:
+        yield
+    except _NET_ERRORS as exc:
+        raise OSError(errno.ENOTCONN, f"SFTP-Verbindung unterbrochen: {exc}") from exc
+
+
+class Source(ABC):
+    kind: str
+
+    def __init__(self, subdir: str = "") -> None:
+        self.subdir = subdir.strip("/")
+
+    def full(self, rel: str) -> str:
+        """Pfad relativ zur Wurzel der Anbindung."""
+        return _join(self.subdir, rel)
+
+    @abstractmethod
+    def describe(self, rel: str = "") -> str: ...
+
+    @abstractmethod
+    def check_root(self) -> None:
+        """Wirft :class:`SourceMissing`, wenn der Quellordner fehlt."""
+
+    @abstractmethod
+    def scan(
+        self, matcher: Matcher, recursive: bool, exclude: frozenset[str] = frozenset()
+    ) -> list[Candidate]:
+        """``exclude``: Verzeichnisse relativ zur Wurzel der Anbindung."""
+
+    @abstractmethod
+    def state(self, rel: str) -> FileState | None:
+        """Frischer Zustand; ``None`` = verschwunden, ``BUSY`` = gerade gesperrt."""
+
+    @abstractmethod
+    def stat(self, rel: str) -> SourceStat: ...
+
+    @abstractmethod
+    def open(self, rel: str) -> contextlib.AbstractContextManager[IO[bytes]]: ...
+
+    @abstractmethod
+    def delete(self, rel: str) -> None: ...
+
+    @abstractmethod
+    def exists_root_rel(self, root_rel: str) -> bool: ...
+
+    @abstractmethod
+    def describe_root_rel(self, root_rel: str) -> str: ...
+
+    @abstractmethod
+    def move_to(self, rel: str, root_rel_target: str) -> None:
+        """Innerhalb der Anbindung verschieben, ohne zu überschreiben
+        (``FileExistsError``, wenn das Ziel existiert)."""
+
+    def local_path(self, rel: str) -> Path | None:
+        """Lokaler Pfad, falls vorhanden (für den Schnellweg ohne Kopie)."""
+        return None
+
+    def close(self) -> None:  # noqa: B027 – absichtlich optional
+        pass
+
+
+class LocalSource(Source):
+    """Eingehängte Freigabe (oder lokales Verzeichnis)."""
+
+    kind = "local"
+
+    def __init__(self, root: Path, subdir: str = "") -> None:
+        super().__init__(subdir)
+        self.root = root
+        self.base = root / self.subdir if self.subdir else root
+        # zusätzlich auszulassende Verzeichnisse, z. B. ein Ziel innerhalb der Quelle
+        self.exclude_paths: set[Path] = set()
+
+    def _p(self, rel: str) -> Path:
+        return self.base / rel
+
+    def describe(self, rel: str = "") -> str:
+        return str(self._p(rel)) if rel else str(self.base)
+
+    def check_root(self) -> None:
+        if not self.base.is_dir():
+            raise SourceMissing(f"Quellordner {self.base} existiert nicht")
+
+    def scan(
+        self, matcher: Matcher, recursive: bool, exclude: frozenset[str] = frozenset()
+    ) -> list[Candidate]:
+        paths = {(self.root / e).resolve() for e in exclude} | set(self.exclude_paths)
+        return list(scan(self.base, matcher, recursive, frozenset(paths)))
+
+    def state(self, rel: str) -> FileState | None:
+        return current_state(self._p(rel))
+
+    def stat(self, rel: str) -> SourceStat:
+        st = os.stat(self._p(rel))
+        return SourceStat(st.st_size, st.st_mtime_ns, st.st_atime_ns, stat.S_IMODE(st.st_mode))
+
+    def open(self, rel: str) -> contextlib.AbstractContextManager[IO[bytes]]:
+        return self._p(rel).open("rb")
+
+    def delete(self, rel: str) -> None:
+        os.unlink(self._p(rel))
+
+    def exists_root_rel(self, root_rel: str) -> bool:
+        return os.path.lexists(self.root / root_rel)
+
+    def describe_root_rel(self, root_rel: str) -> str:
+        return str(self.root / root_rel)
+
+    def move_to(self, rel: str, root_rel_target: str) -> None:
+        target = self.root / root_rel_target
+        target.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
+        if rename_noreplace(self._p(rel), target):
+            return
+        if os.path.lexists(target):
+            raise FileExistsError(errno.EEXIST, "exists", str(target))
+        os.rename(self._p(rel), target)
+
+    def local_path(self, rel: str) -> Path:
+        return self._p(rel)
+
+
+class SftpSource(Source):
+    """Quelle auf einem SFTP-Server (z. B. ein Server, der nur per SSH erreichbar ist)."""
+
+    kind = "sftp"
+
+    def __init__(self, session: SftpSession, base: str, subdir: str = "", label: str = ""):
+        super().__init__(subdir)
+        self.session = session
+        self.client = session.client
+        self.root = base.rstrip("/") or "/"
+        self.label = label or "sftp"
+
+    def _p(self, rel: str) -> str:
+        return posixpath.join(self.root, self.full(rel)) if self.full(rel) else self.root
+
+    def describe(self, rel: str = "") -> str:
+        return f"{self.label}:{self._p(rel)}"
+
+    def close(self) -> None:
+        self.session.close()
+
+    def check_root(self) -> None:
+        with network_errors():
+            try:
+                attrs = self.client.stat(self._p(""))
+            except FileNotFoundError:
+                raise SourceMissing(f"Quellordner {self.describe()} existiert nicht") from None
+        if attrs.st_mode is None or not stat.S_ISDIR(attrs.st_mode):
+            raise SourceMissing(f"{self.describe()} ist kein Verzeichnis")
+
+    def scan(
+        self, matcher: Matcher, recursive: bool, exclude: frozenset[str] = frozenset()
+    ) -> list[Candidate]:
+        excluded = {e.strip("/") for e in exclude}
+        found: list[Candidate] = []
+
+        def walk(rel_dir: str) -> None:
+            with network_errors():
+                entries = sorted(
+                    self.client.listdir_attr(self._p(rel_dir)), key=lambda a: a.filename
+                )
+            for a in entries:
+                name = a.filename
+                if name.startswith(".") or a.st_mode is None:
+                    continue
+                rel = _join(rel_dir, name)
+                if stat.S_ISDIR(a.st_mode):
+                    if recursive and name not in IGNORED_DIRS and self.full(rel) not in excluded:
+                        walk(rel)
+                    continue
+                if not stat.S_ISREG(a.st_mode) or is_ignored_name(name):
+                    continue  # Symlinks, Geräte … und unfertige Übertragungen
+                if matcher(rel):
+                    found.append(Candidate(rel, None, _state(a)))
+
+        walk("")
+        return found
+
+    def state(self, rel: str) -> FileState | None:
+        with network_errors():
+            try:
+                a = self.client.lstat(self._p(rel))
+            except FileNotFoundError:
+                return None
+        if a.st_mode is None or not stat.S_ISREG(a.st_mode):
+            return None
+        return _state(a)
+
+    def stat(self, rel: str) -> SourceStat:
+        with network_errors():
+            a = self.client.lstat(self._p(rel))
+        mtime = int(a.st_mtime or 0) * 1_000_000_000
+        atime = int(a.st_atime or a.st_mtime or 0) * 1_000_000_000
+        return SourceStat(int(a.st_size or 0), mtime, atime, stat.S_IMODE(a.st_mode or 0o644))
+
+    @contextlib.contextmanager
+    def _reader(self, rel: str) -> Iterator[IO[bytes]]:
+        with network_errors():
+            size = self.client.lstat(self._p(rel)).st_size
+            fh: Any = self.client.open(self._p(rel), "rb")
+            try:
+                fh.prefetch(size)  # Größe mitgeben: spart ein fstat und liest vorab
+                yield fh
+            finally:
+                with contextlib.suppress(*_NET_ERRORS, OSError):
+                    fh.close()
+
+    def open(self, rel: str) -> contextlib.AbstractContextManager[IO[bytes]]:
+        return self._reader(rel)
+
+    def delete(self, rel: str) -> None:
+        with network_errors():
+            self.client.remove(self._p(rel))
+
+    def exists_root_rel(self, root_rel: str) -> bool:
+        with network_errors():
+            try:
+                self.client.lstat(posixpath.join(self.root, root_rel))
+            except FileNotFoundError:
+                return False
+        return True
+
+    def describe_root_rel(self, root_rel: str) -> str:
+        return f"{self.label}:{posixpath.join(self.root, root_rel)}"
+
+    def move_to(self, rel: str, root_rel_target: str) -> None:
+        target = posixpath.join(self.root, root_rel_target)
+        with network_errors():
+            current = self.root
+            for part in posixpath.dirname(root_rel_target).split("/"):
+                if not part:
+                    continue
+                current = posixpath.join(current, part)
+                try:
+                    self.client.stat(current)
+                except FileNotFoundError:
+                    self.client.mkdir(current, mode=0o755)
+            if self.exists_root_rel(root_rel_target):
+                raise FileExistsError(errno.EEXIST, "exists", target)
+            self.client.rename(self._p(rel), target)  # SFTPv3-rename überschreibt nicht
+
+
+def _state(a: paramiko.SFTPAttributes) -> FileState:
+    return FileState(int(a.st_size or 0), int(a.st_mtime or 0) * 1_000_000_000)
diff --git a/tests/sftp_stub.py b/tests/sftp_stub.py
index 8385f63..8d8f52e 100644
--- a/tests/sftp_stub.py
+++ b/tests/sftp_stub.py
@@ -48,7 +48,11 @@ class _Server(paramiko.ServerInterface):
 
 
 class _Handle(SFTPHandle):
-    pass
+    def stat(self) -> Any:  # fstat wie bei OpenSSH
+        try:
+            return SFTPAttributes.from_stat(os.fstat(self.readfile.fileno()))
+        except OSError as e:
+            return SFTPServer.convert_errno(e.errno)
 
 
 class _RootedSFTP(SFTPServerInterface):
diff --git a/tests/test_engine.py b/tests/test_engine.py
index a2e50f3..17c4001 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -251,7 +251,7 @@ def test_source_not_deletable_keeps_target_and_prevents_duplicate(
             raise PermissionError(errno.EACCES, "Permission denied", str(path))
         real_unlink(path)  # type: ignore[arg-type]
 
-    monkeypatch.setattr("netmover.mover.engine.os.unlink", deny)
+    monkeypatch.setattr("netmover.mover.sources.os.unlink", deny)
     first = run(engine, snap(rule_id), src, dst)
     assert first.items[0].status is ItemStatus.FAILED
     assert "nicht löschbar" in (first.items[0].message or "")
diff --git a/tests/test_sftp_source.py b/tests/test_sftp_source.py
new file mode 100644
index 0000000..b7cbe3e
--- /dev/null
+++ b/tests/test_sftp_source.py
@@ -0,0 +1,398 @@
+"""0.5.0: SFTP als Quelle – dieselben Garantien wie bei eingehängten Freigaben."""
+
+from __future__ import annotations
+
+import errno
+import hashlib
+import os
+from collections.abc import Callable, Iterator
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from netmover.crypto import SecretBox
+from netmover.db import session_scope
+from netmover.models import (
+    CopiedFile,
+    Credential,
+    CredentialKind,
+    ItemStatus,
+    Mount,
+    MountType,
+    PatternType,
+    Rule,
+    RuleAction,
+    RunStatus,
+)
+from netmover.mounting.manager import MountManager
+from netmover.mounting.sftp import SftpConfig, connect
+from netmover.mover.destinations import LocalDestination, SftpDestination
+from netmover.mover.engine import Engine, RuleSnapshot, RunResult
+from netmover.mover.ledger import Ledger
+from netmover.mover.runner import RuleRunner
+from netmover.mover.scan import Matcher
+from netmover.mover.sources import SftpSource
+from netmover.timeouts import CancelToken
+from tests.conftest import MountFactory
+from tests.sftp_stub import StubSFTPServer
+
+
+@pytest.fixture
+def server(tmp_path: Path) -> Iterator[StubSFTPServer]:
+    root = tmp_path / "sftp-root"
+    for d in ("eingang", "ziel", "consume"):
+        (root / d).mkdir(parents=True)
+    srv = StubSFTPServer(root).start()
+    yield srv
+    srv.stop()
+
+
+def _cfg(server: StubSFTPServer) -> SftpConfig:
+    return SftpConfig(
+        "127.0.0.1",
+        server.port,
+        server.username,
+        "/",
+        private_key=server.client_private_key,
+        host_key=server.host_key_string,
+    )
+
+
+@pytest.fixture
+def source(server: StubSFTPServer) -> Iterator[SftpSource]:
+    s = SftpSource(connect(_cfg(server)), "/", "eingang", label="nas2")
+    yield s
+    s.close()
+
+
+@pytest.fixture
+def rule_id(make_mount: MountFactory, session_factory: sessionmaker[Session]) -> int:
+    mid = make_mount()
+    with session_scope(session_factory) as s:
+        r = Rule(name="r", src_mount_id=mid, dst_path="/x")
+        s.add(r)
+        s.flush()
+        return r.id
+
+
+@pytest.fixture
+def hooks() -> list[Callable[[], None]]:
+    return []
+
+
+@pytest.fixture
+def engine(session_factory: sessionmaker[Session], hooks: list[Callable[[], None]]) -> Engine:
+    def sleep(_: float) -> None:
+        for h in hooks:
+            h()
+
+    return Engine(Ledger(session_factory), sleep=sleep)
+
+
+def snap(rule_id: int, **kw: object) -> RuleSnapshot:
+    return RuleSnapshot(id=rule_id, name="r", stability_seconds=2, **kw)  # type: ignore[arg-type]
+
+
+def write(path: Path, data: bytes = b"inhalt") -> Path:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(data)
+    return path
+
+
+def run(
+    engine: Engine, rule: RuleSnapshot, src: SftpSource, dest: object, **kw: object
+) -> RunResult:
+    return engine.execute(rule, src, dest, CancelToken(), **kw)  # type: ignore[arg-type]
+
+
+def sha(p: Path) -> str:
+    return hashlib.sha256(p.read_bytes()).hexdigest()
+
+
+# -- Scan ----------------------------------------------------------------------------------
+
+
+def test_scan_filters_like_local_sources(source: SftpSource, server: StubSFTPServer) -> None:
+    root = server.root / "eingang"
+    for name in ("a.pdf", "b.PDF", "notiz.txt", ".versteckt.pdf", "halb.pdf.part", "x.tmp"):
+        write(root / name)
+    write(root / "max" / "c.pdf")
+    write(root / "@eaDir" / "a.pdf" / "thumb.pdf")
+    write(root / "quarantaene" / "d.pdf")
+    m = Matcher("*.pdf", PatternType.GLOB)
+    assert [c.relpath for c in source.scan(m, recursive=False)] == ["a.pdf", "b.PDF"]
+    rec = [
+        c.relpath
+        for c in source.scan(m, recursive=True, exclude=frozenset({"eingang/quarantaene"}))
+    ]
+    assert rec == ["a.pdf", "b.PDF", "max/c.pdf"]
+    c = source.scan(m, recursive=False)[0]
+    assert c.path is None and c.state.size == len(b"inhalt")
+
+
+def test_missing_source_dir_aborts(
+    engine: Engine, rule_id: int, server: StubSFTPServer, tmp_path: Path
+) -> None:
+    src = SftpSource(connect(_cfg(server)), "/", "gibtsnicht")
+    try:
+        result = run(
+            engine, snap(rule_id), src, LocalDestination(tmp_path / "dst", create_root=True)
+        )
+    finally:
+        src.close()
+    assert result.status is RunStatus.ERROR and "existiert nicht" in (result.message or "")
+
+
+# -- Verschieben / Kopieren ----------------------------------------------------------------
+
+
+def test_move_sftp_to_local(
+    engine: Engine, rule_id: int, source: SftpSource, server: StubSFTPServer, tmp_path: Path
+) -> None:
+    f = write(server.root / "eingang" / "scan001.pdf", os.urandom(300_000))
+    digest = sha(f)
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    write(dst / "scan001.pdf", b"alt")
+    result = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert result.status is RunStatus.OK, result.message
+    assert [(i.src, i.status) for i in result.items] == [("scan001.pdf", ItemStatus.MOVED)]
+    assert sha(dst / "scan001-1.pdf") == digest and (dst / "scan001.pdf").read_bytes() == b"alt"
+    assert not f.exists()
+    assert sorted(p.name for p in dst.iterdir()) == ["scan001-1.pdf", "scan001.pdf"]
+
+
+def test_move_sftp_to_sftp(
+    engine: Engine, rule_id: int, source: SftpSource, server: StubSFTPServer
+) -> None:
+    f = write(server.root / "eingang" / "a.pdf", os.urandom(50_000))
+    digest = sha(f)
+    dest = SftpDestination(connect(_cfg(server)), "/ziel", label="nas3")
+    try:
+        result = run(engine, snap(rule_id), source, dest)
+    finally:
+        dest.close()
+    assert result.items[0].status is ItemStatus.MOVED and result.items[0].dst == "nas3:/ziel/a.pdf"
+    assert sha(server.root / "ziel" / "a.pdf") == digest and not f.exists()
+    assert list((server.root / "ziel").glob(".netmover-*")) == []
+
+
+def test_copy_mode_remembers_versions(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    session_factory: sessionmaker[Session],
+) -> None:
+    f = write(server.root / "eingang" / "a.pdf", b"v1")
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    rule = snap(rule_id, action=RuleAction.COPY)
+    for _ in range(3):
+        run(engine, rule, source, LocalDestination(dst))
+    assert sorted(p.name for p in dst.iterdir()) == ["a.pdf"] and f.exists()
+    f.write_bytes(b"version 2")  # andere Größe → neue Version
+    run(engine, rule, source, LocalDestination(dst))
+    assert sorted(p.name for p in dst.iterdir()) == ["a-1.pdf", "a.pdf"]
+    with session_factory() as s:
+        assert len(s.scalars(select(CopiedFile)).all()) == 2
+
+
+def test_mtime_is_carried_over(
+    engine: Engine, rule_id: int, source: SftpSource, server: StubSFTPServer, tmp_path: Path
+) -> None:
+    f = write(server.root / "eingang" / "a.pdf")
+    os.utime(f, (1_600_000_000, 1_600_000_123))
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    run(engine, snap(rule_id, action=RuleAction.COPY), source, LocalDestination(dst))
+    assert int((dst / "a.pdf").stat().st_mtime) == 1_600_000_123
+
+
+# -- Fallstricke ---------------------------------------------------------------------------
+
+
+def test_growing_file_is_not_touched(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    hooks: list[Callable[[], None]],
+) -> None:
+    growing = write(server.root / "eingang" / "wachsend.pdf", b"x" * 10)
+    hooks.append(lambda: growing.write_bytes(b"x" * 20))
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    result = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert result.items[0].status is ItemStatus.SKIPPED and growing.exists()
+    assert list(dst.iterdir()) == []
+
+
+def test_source_changed_during_transfer_keeps_source(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    f = write(server.root / "eingang" / "a.pdf", b"original")
+    real = LocalDestination._apply_meta
+
+    def change(self: LocalDestination, rel: str, meta: object) -> list[str]:
+        f.write_bytes(b"geaendert!!")
+        return real(self, rel, meta)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(LocalDestination, "_apply_meta", change)
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    result = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert result.items[0].status is ItemStatus.SKIPPED and f.exists()
+    assert list(dst.iterdir()) == []
+
+
+def test_source_not_deletable_prevents_duplicates(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    write(server.root / "eingang" / "a.pdf")
+
+    def deny(*a: object, **kw: object) -> None:
+        raise PermissionError(errno.EACCES, "Permission denied")
+
+    monkeypatch.setattr(source.client, "remove", deny)
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    first = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert first.items[0].status is ItemStatus.FAILED and "nicht löschbar" in (
+        first.items[0].message or ""
+    )
+    second = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert second.items == [] and sorted(p.name for p in dst.iterdir()) == ["a.pdf"]
+
+
+def test_quarantine_on_sftp_source(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from netmover.mover.destinations import TransferError
+
+    def broken(*a: object, **kw: object) -> None:
+        raise TransferError("kaputt")
+
+    monkeypatch.setattr(LocalDestination, "_link_noreplace", broken)
+    write(server.root / "eingang" / "defekt.pdf")
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    rule = snap(rule_id, quarantine_after=2)
+    for _ in range(2):
+        run(engine, rule, source, LocalDestination(dst), quarantine_rel="eingang/_q")
+    r = run(engine, rule, source, LocalDestination(dst), quarantine_rel="eingang/_q")
+    assert r.items[0].status is ItemStatus.QUARANTINED
+    assert r.items[0].dst == "nas2:/eingang/_q/defekt.pdf"
+    assert (server.root / "eingang" / "_q" / "defekt.pdf").exists()
+
+
+def test_connection_loss_aborts_run(
+    engine: Engine,
+    rule_id: int,
+    source: SftpSource,
+    server: StubSFTPServer,
+    tmp_path: Path,
+    hooks: list[Callable[[], None]],
+) -> None:
+    write(server.root / "eingang" / "a.pdf")
+    write(server.root / "eingang" / "b.pdf")
+    hooks.append(lambda: source.session.transport.close())  # Verbindung bricht weg
+    dst = tmp_path / "dst"
+    dst.mkdir()
+    result = run(engine, snap(rule_id), source, LocalDestination(dst))
+    assert result.status is RunStatus.ERROR
+    assert (server.root / "eingang" / "a.pdf").exists() and list(dst.iterdir()) == []
+
+
+# -- Regellauf mit SFTP-Quelle ------------------------------------------------------------
+
+
+def test_runner_with_sftp_source(
+    server: StubSFTPServer,
+    manager: MountManager,
+    session_factory: sessionmaker[Session],
+    secretbox: SecretBox,
+    tmp_path: Path,
+) -> None:
+    write(server.root / "eingang" / "scan.pdf", b"%PDF")
+    consume = tmp_path / "consume"
+    consume.mkdir()
+    with session_scope(session_factory) as s:
+        c = Credential(
+            name="k",
+            kind=CredentialKind.SSH_KEY,
+            username=server.username,
+            secret_enc=secretbox.encrypt(server.client_private_key),
+        )
+        s.add(c)
+        s.flush()
+        m = Mount(
+            name="Scanner-Server",
+            slug="scanner",
+            type=MountType.SFTP,
+            server="127.0.0.1",
+            share="/",
+            port=server.port,
+            cred_id=c.id,
+            host_key=server.host_key_string,
+        )
+        s.add(m)
+        s.flush()
+        r = Rule(
+            name="SFTP → lokal",
+            src_mount_id=m.id,
+            src_subpath="eingang",
+            dst_path=str(consume),
+            stability_seconds=2,
+        )
+        s.add(r)
+        s.flush()
+        rid = r.id
+    result = RuleRunner(session_factory, manager, sleep=lambda _: None).run(rid)
+    assert result.status is RunStatus.OK, result.message
+    assert result.items[0].dst == str(consume / "scan.pdf")
+    assert (consume / "scan.pdf").read_bytes() == b"%PDF"
+    assert not (server.root / "eingang" / "scan.pdf").exists()
+
+
+def test_rule_form_accepts_sftp_source_with_min_stability(
+    session_factory: sessionmaker[Session],
+) -> None:
+    from netmover.admin import parse_rule_form
+
+    with session_scope(session_factory) as s:
+        m = Mount(name="sftp", slug="sftp", type=MountType.SFTP, server="10.0.0.1", share="/in")
+        s.add(m)
+        s.flush()
+        form = {
+            "name": "x",
+            "src_mount_id": str(m.id),
+            "dst_kind": "local",
+            "dst_path": "/tmp",
+            "interval_seconds": "30",
+            "stability_seconds": "1",
+            "op_timeout_seconds": "120",
+        }
+        _, errors = parse_rule_form(s, form)
+        assert "src_mount_id" not in errors and "mindestens 2 s" in errors["stability_seconds"]
+        _, errors = parse_rule_form(s, {**form, "stability_seconds": "3"})
+        assert errors == {}