Add SFTP, FTP and SMB network sources to the processor (#7153)

# Description of Changes

Add SFTP, FTP and SMB network sources to the processor plus UI change to
enable it

---

## Checklist

### General

- [ ] I have read the [Contribution
Guidelines](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/CONTRIBUTING.md)
- [ ] I have read the [Stirling-PDF Developer
Guide](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/DeveloperGuide.md)
(if applicable)
- [ ] I have read the [How to add new languages to
Stirling-PDF](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/devGuide/HowToAddNewLanguage.md)
(if applicable)
- [ ] I have performed a self-review of my own code
- [ ] My changes generate no new warnings

### Documentation

- [ ] I have updated relevant docs on [Stirling-PDF's doc
repo](https://github.com/Stirling-Tools/Stirling-Tools.github.io/blob/main/docs/)
(if functionality has heavily changed)
- [ ] I have read the section [Add New Translation
Tags](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/devGuide/HowToAddNewLanguage.md#add-new-translation-tags)
(for new translation tags only)

### Translations (if applicable)

- [ ] I ran
[`scripts/counter_translation.py`](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/docs/counter_translation.md)

### UI Changes (if applicable)

- [ ] Screenshots or videos demonstrating the UI changes are attached
(e.g., as comments or direct attachments in the PR)

### Testing (if applicable)

- [ ] I have run `task check` to verify linters, typechecks, and tests
pass
- [ ] I have tested my changes locally. Refer to the [Testing
Guide](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/DeveloperGuide.md#7-testing)
for more details.

app/allowed-licenses.json

@@ -32,6 +32,14 @@
       "moduleName": ".*",
       "moduleLicense": "BSD-4 License"
     },
+    {
+      "moduleName": ".*",
+      "moduleLicense": "Revised BSD"
+    },
+    {
+      "moduleName": ".*",
+      "moduleLicense": "ISC"
+    },
     {
       "moduleName": ".*",
       "moduleLicense": "MIT"
@@ -48,6 +56,10 @@
       "moduleName": ".*",
       "moduleLicense": "MIT-0"
     },
+    {
+      "moduleName": ".*",
+      "moduleLicense": "MIT license"
+    },
     {
       "moduleName": "com.github.jai-imageio:jai-imageio-core",
       "moduleLicense": "LICENSE.txt"

app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java

@@ -251,6 +251,21 @@ public class ApplicationProperties {
          */
         private boolean allowPrivateS3Endpoints = false;
 
+        /**
+         * Whether a network source's host (SFTP, FTP, or SMB) may resolve to a loopback,
+         * link-local, or private address. Off by default so a connection cannot be pointed at
+         * internal services; enable for an on-network file server (e.g. an internal SFTP drop or a
+         * Samba share).
+         */
+        private boolean allowPrivateNetworkSources = false;
+
+        /**
+         * Hostnames (exact, case-insensitive) that a network source may use even when they resolve
+         * to a private or local address and {@code allowPrivateNetworkSources} is off. Lets shared
+         * infra allow one named on-prem file server without opening every internal host.
+         */
+        private List<String> allowedPrivateNetworkHosts = new java.util.ArrayList<>();
+
         /**
          * Whether an API/Purview/ConsignO integration's base URL may resolve to a loopback,
          * link-local, or private address. Off by default: unlike S3 connections, any user may

app/proprietary/build.gradle

@@ -67,6 +67,12 @@ dependencies {
     implementation "software.amazon.awssdk:s3:${awsSdkVersion}"
     implementation "software.amazon.awssdk:url-connection-client:${awsSdkVersion}"
 
+    // Network policy sources (SFTP/FTP/SMB drop folders). All permissively licensed:
+    // jsch fork BSD, commons-net Apache-2.0, smbj Apache-2.0 (BouncyCastle already on classpath).
+    implementation "com.github.mwiede:jsch:${jschVersion}"
+    implementation "commons-net:commons-net:${commonsNetVersion}"
+    implementation "com.hierynomus:smbj:${smbjVersion}"
+
     // Streaming AEAD (AES-GCM-HKDF segments) for storage encryption at rest. Apache-2.0.
     implementation "com.google.crypto.tink:tink:${tinkVersion}"
 

app/proprietary/src/main/java/stirling/software/proprietary/integration/model/IntegrationType.java

@@ -3,6 +3,8 @@ package stirling.software.proprietary.integration.model;
 /** Kind of external integration a stored config describes. */
 public enum IntegrationType {
     S3,
+    /** A network file server (SFTP, FTP/FTPS, or SMB/Samba) a policy source polls for documents. */
+    NETWORK,
     MCP,
     /** A generic outbound HTTP endpoint a pipeline step can post a document to. */
     API,

app/proprietary/src/main/java/stirling/software/proprietary/integration/service/IntegrationConfigService.java

@@ -57,12 +57,15 @@ public class IntegrationConfigService {
         OwnerScope scope = request.scope() == null ? OwnerScope.USER : request.scope();
         IntegrationConfig cfg = new IntegrationConfig();
         cfg.setIntegrationType(require(request.integrationType(), "integrationType"));
-        // S3 is infrastructure, not self-serve: no personal S3 for regular users. TEAM/SERVER
-        // scopes are already restricted to admins/team owners by assignOwnership.
-        if (cfg.getIntegrationType() == IntegrationType.S3
+        // S3 and network file servers are infrastructure, not self-serve: they hold shared host
+        // credentials, so no personal connection for regular users. TEAM/SERVER scopes are already
+        // restricted to admins/team owners by assignOwnership.
+        if (isInfrastructureType(cfg.getIntegrationType())
                 && scope == OwnerScope.USER
                 && !ownership.isAdmin(currentUser)) {
-            throw forbidden("S3 connections can only be created by administrators or team owners");
+            throw forbidden(
+                    "S3 and network connections can only be created by administrators or team"
+                            + " owners");
         }
         requireCustomApiAllowed(cfg.getIntegrationType(), currentUser);
         cfg.setName(require(request.name(), "name"));
@@ -147,6 +150,11 @@ public class IntegrationConfigService {
         }
     }
 
+    /** Types holding shared host credentials, restricted to admins/team owners like S3. */
+    private static boolean isInfrastructureType(IntegrationType type) {
+        return type == IntegrationType.S3 || type == IntegrationType.NETWORK;
+    }
+
     /** Whether this caller may author custom API integrations, for the UI to offer or hide it. */
     public boolean canAuthorCustomApi(User currentUser) {
         return applicationProperties.getPolicies().isAllowCustomApiIntegrations()

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/FtpFileClient.java

@@ -0,0 +1,181 @@
+package stirling.software.proprietary.policy.network;
+
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.net.ftp.FTP;
+import org.apache.commons.net.ftp.FTPClient;
+import org.apache.commons.net.ftp.FTPFile;
+import org.apache.commons.net.ftp.FTPReply;
+import org.apache.commons.net.ftp.FTPSClient;
+
+/**
+ * FTP client over Apache Commons Net, with optional TLS ({@code EXPLICIT} = AUTH TLS, {@code
+ * IMPLICIT} = TLS on connect). Passive mode is the default since it works through NAT. A blank
+ * directory lists the login working directory. One transfer is in flight at a time, so the read
+ * stream calls {@code completePendingCommand} on close before the session is reused or shut down.
+ */
+final class FtpFileClient implements RemoteFileClient {
+
+    private static final int CONNECT_TIMEOUT_MS = 15_000;
+    private static final int READ_TIMEOUT_MS = 60_000;
+    private static final int MAX_DEPTH = 64;
+
+    private final FTPClient ftp;
+
+    private FtpFileClient(FTPClient ftp) {
+        this.ftp = ftp;
+    }
+
+    static FtpFileClient connect(NetworkConfig config) throws IOException {
+        FTPClient ftp = clientFor(config);
+        ftp.setConnectTimeout(CONNECT_TIMEOUT_MS);
+        // Read timeouts so a server that connects then stalls cannot hang a poller: default
+        // becomes the control socket's SO timeout at connect, data covers transfer sockets.
+        ftp.setDefaultTimeout(READ_TIMEOUT_MS);
+        ftp.setDataTimeout(Duration.ofMillis(READ_TIMEOUT_MS));
+        try {
+            ftp.connect(config.host(), config.port());
+            int reply = ftp.getReplyCode();
+            if (!FTPReply.isPositiveCompletion(reply)) {
+                ftp.disconnect();
+                throw new IOException(
+                        "FTP server " + config.host() + " refused connection: " + reply);
+            }
+            if (ftp instanceof FTPSClient ftps) {
+                // Encrypt the data channel too, not only the control channel.
+                ftps.execPBSZ(0);
+                ftps.execPROT("P");
+            }
+            if (!ftp.login(config.username(), config.password())) {
+                int code = ftp.getReplyCode();
+                ftp.disconnect();
+                throw new IOException("FTP login to " + config.host() + " failed: " + code);
+            }
+            if (config.passive()) {
+                ftp.enterLocalPassiveMode();
+            }
+            ftp.setFileType(FTP.BINARY_FILE_TYPE);
+            return new FtpFileClient(ftp);
+        } catch (IOException e) {
+            quietlyDisconnect(ftp);
+            throw e;
+        }
+    }
+
+    private static FTPClient clientFor(NetworkConfig config) {
+        return switch (config.security()) {
+            case NONE -> new FTPClient();
+            case IMPLICIT -> new FTPSClient(true);
+            case EXPLICIT -> new FTPSClient(false);
+        };
+    }
+
+    @Override
+    public List<RemoteFile> list(String directory, boolean recursive) throws IOException {
+        List<RemoteFile> files = new ArrayList<>();
+        collect(dir(directory), recursive, 0, files);
+        return files;
+    }
+
+    private void collect(String directory, boolean recursive, int depth, List<RemoteFile> out)
+            throws IOException {
+        FTPFile[] entries = ftp.listFiles(directory);
+        for (FTPFile entry : entries) {
+            if (out.size() >= MAX_FILES) {
+                return;
+            }
+            if (entry == null) {
+                continue;
+            }
+            String name = entry.getName();
+            if (name.equals(".") || name.equals("..") || name.startsWith(".")) {
+                continue;
+            }
+            String path = join(directory, name);
+            if (entry.isDirectory()) {
+                if (recursive && !entry.isSymbolicLink() && depth < MAX_DEPTH) {
+                    collect(path, true, depth + 1, out);
+                }
+                continue;
+            }
+            if (entry.isFile()) {
+                out.add(new RemoteFile(path, name, entry.getSize(), lastModified(entry)));
+            }
+        }
+    }
+
+    @Override
+    public RemoteFile stat(String path) throws IOException {
+        // LIST, not MLST: listing uses listFiles, so the re-stat that guards the consume delete
+        // must read the timestamp the same way or the version gate never matches (and many servers
+        // do not implement MLST at all, returning null and skipping the delete entirely).
+        FTPFile[] listed = ftp.listFiles(path);
+        if (listed.length == 0 || listed[0] == null) {
+            return null;
+        }
+        FTPFile entry = listed[0];
+        String name = path.substring(path.lastIndexOf('/') + 1);
+        return new RemoteFile(path, name, entry.getSize(), lastModified(entry));
+    }
+
+    @Override
+    public InputStream open(String path) throws IOException {
+        InputStream stream = ftp.retrieveFileStream(path);
+        if (stream == null) {
+            throw new IOException("cannot read " + path + ": " + ftp.getReplyString());
+        }
+        // The data stream must be closed and the transfer acknowledged before the session is
+        // reused.
+        return new FilterInputStream(stream) {
+            @Override
+            public void close() throws IOException {
+                super.close();
+                ftp.completePendingCommand();
+            }
+        };
+    }
+
+    @Override
+    public void delete(String path) throws IOException {
+        ftp.deleteFile(path);
+    }
+
+    @Override
+    public void close() throws IOException {
+        try {
+            ftp.logout();
+        } finally {
+            quietlyDisconnect(ftp);
+        }
+    }
+
+    private static long lastModified(FTPFile entry) {
+        return entry.getTimestamp() == null ? 0L : entry.getTimestamp().getTimeInMillis();
+    }
+
+    private static void quietlyDisconnect(FTPClient ftp) {
+        if (ftp.isConnected()) {
+            try {
+                ftp.disconnect();
+            } catch (IOException ignored) {
+                // Best-effort teardown; the session is being abandoned anyway.
+            }
+        }
+    }
+
+    private static String dir(String directory) {
+        return directory == null || directory.isBlank() ? "." : directory;
+    }
+
+    private static String join(String directory, String name) {
+        if (directory.equals(".")) {
+            return name;
+        }
+        return directory.endsWith("/") ? directory + name : directory + "/" + name;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/NetworkConfig.java

@@ -0,0 +1,222 @@
+package stirling.software.proprietary.policy.network;
+
+import java.util.Map;
+
+/**
+ * The fully resolved settings a network source runs with - produced by {@link
+ * NetworkConnectionResolver} merging a stored connection (protocol, host, credentials, and, for
+ * SMB, the share) with per-use source options (directory, mode, recursive). Credentials are
+ * required: there is no anonymous fallback, so a source can never reach a server without
+ * config-supplied identity. {@code snapshot} and {@code recursive} are input-only.
+ */
+public record NetworkConfig(
+        NetworkProtocol protocol,
+        String host,
+        int port,
+        String username,
+        String password,
+        String privateKey,
+        String privateKeyPassphrase,
+        String hostKeyFingerprint,
+        String domain,
+        String share,
+        FtpSecurity security,
+        boolean passive,
+        String directory,
+        boolean snapshot,
+        boolean recursive) {
+
+    /** FTP transport security: plaintext, or TLS negotiated up-front (implicit) or via AUTH TLS. */
+    public enum FtpSecurity {
+        NONE,
+        EXPLICIT,
+        IMPLICIT;
+
+        static FtpSecurity fromOption(String value) {
+            if (value == null || value.isBlank()) {
+                return NONE;
+            }
+            String normalized = value.trim().toUpperCase();
+            for (FtpSecurity security : values()) {
+                if (security.name().equals(normalized)) {
+                    return security;
+                }
+            }
+            throw new IllegalArgumentException(
+                    "network config 'security' must be NONE, EXPLICIT or IMPLICIT");
+        }
+    }
+
+    private static final String PROTOCOL_OPTION = "protocol";
+    private static final String HOST_OPTION = "host";
+    private static final String PORT_OPTION = "port";
+    private static final String USERNAME_OPTION = "username";
+    private static final String PASSWORD_OPTION = "password";
+    private static final String PRIVATE_KEY_OPTION = "privateKey";
+    private static final String PRIVATE_KEY_PASSPHRASE_OPTION = "privateKeyPassphrase";
+    private static final String HOST_KEY_FINGERPRINT_OPTION = "hostKeyFingerprint";
+    private static final String DOMAIN_OPTION = "domain";
+    private static final String SHARE_OPTION = "share";
+    private static final String SECURITY_OPTION = "security";
+    private static final String PASSIVE_OPTION = "passive";
+    private static final String DIRECTORY_OPTION = "directory";
+    private static final String MODE_OPTION = "mode";
+    private static final String MODE_CONSUME = "consume";
+    private static final String MODE_SNAPSHOT = "snapshot";
+    private static final String RECURSIVE_OPTION = "recursive";
+
+    public static NetworkConfig from(Map<String, Object> options) {
+        NetworkProtocol protocol = NetworkProtocol.fromOption(str(options.get(PROTOCOL_OPTION)));
+        if (protocol == null) {
+            throw new IllegalArgumentException(
+                    "network config requires a 'protocol' of sftp, ftp or smb");
+        }
+        String host = trimmed(options.get(HOST_OPTION));
+        if (host == null) {
+            throw new IllegalArgumentException("network config requires a 'host'");
+        }
+        FtpSecurity security = FtpSecurity.fromOption(str(options.get(SECURITY_OPTION)));
+        int port = port(options.get(PORT_OPTION), defaultPort(protocol, security));
+        String username = trimmed(options.get(USERNAME_OPTION));
+        if (username == null) {
+            throw new IllegalArgumentException("network config requires a 'username'");
+        }
+        String password = trimmed(options.get(PASSWORD_OPTION));
+        String privateKey = trimmed(options.get(PRIVATE_KEY_OPTION));
+        String passphrase = trimmed(options.get(PRIVATE_KEY_PASSPHRASE_OPTION));
+        String fingerprint = trimmed(options.get(HOST_KEY_FINGERPRINT_OPTION));
+        String domain = trimmed(options.get(DOMAIN_OPTION));
+        String share = trimmed(options.get(SHARE_OPTION));
+        boolean passive = parseBoolean(options.get(PASSIVE_OPTION), true);
+        String directory = directory(options.get(DIRECTORY_OPTION));
+        boolean recursive = parseBoolean(options.get(RECURSIVE_OPTION), false);
+        boolean snapshot = snapshot(str(options.get(MODE_OPTION)));
+
+        if (protocol == NetworkProtocol.SFTP && password == null && privateKey == null) {
+            throw new IllegalArgumentException(
+                    "sftp connection requires a 'password' or a 'privateKey'");
+        }
+        if ((protocol == NetworkProtocol.FTP || protocol == NetworkProtocol.SMB)
+                && password == null) {
+            throw new IllegalArgumentException(
+                    protocol.name() + " connection requires a 'password'");
+        }
+        if (protocol == NetworkProtocol.SMB && share == null) {
+            throw new IllegalArgumentException(
+                    "smb connection requires a 'share' (e.g. documents)");
+        }
+        return new NetworkConfig(
+                protocol,
+                host,
+                port,
+                username,
+                password,
+                privateKey,
+                passphrase,
+                fingerprint,
+                domain,
+                share,
+                security,
+                passive,
+                directory,
+                snapshot,
+                recursive);
+    }
+
+    /** Implicit FTPS listens on 990, not the plain-FTP 21. */
+    private static int defaultPort(NetworkProtocol protocol, FtpSecurity security) {
+        if (protocol == NetworkProtocol.FTP && security == FtpSecurity.IMPLICIT) {
+            return 990;
+        }
+        return protocol.defaultPort();
+    }
+
+    private static int port(Object value, int fallback) {
+        String text = trimmed(value);
+        if (text == null) {
+            return fallback;
+        }
+        int port;
+        try {
+            port = Integer.parseInt(text);
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException("network config 'port' must be a number");
+        }
+        if (port < 1 || port > 65535) {
+            throw new IllegalArgumentException("network config 'port' must be between 1 and 65535");
+        }
+        return port;
+    }
+
+    /**
+     * A poll directory relative to the login/home dir (SFTP/FTP) or the share root (SMB). Traversal
+     * segments are rejected so a connection cannot be steered above where the operator scoped it.
+     */
+    private static String directory(Object value) {
+        String text = trimmed(value);
+        if (text == null) {
+            return "";
+        }
+        String normalized = text.replace('\\', '/');
+        if (normalized.indexOf('\0') >= 0) {
+            throw new IllegalArgumentException("network config 'directory' contains a null byte");
+        }
+        for (String segment : normalized.split("/")) {
+            if (segment.equals("..")) {
+                throw new IllegalArgumentException(
+                        "network config 'directory' must not contain '..'");
+            }
+        }
+        return normalized;
+    }
+
+    private static boolean snapshot(String mode) {
+        if (mode == null || mode.isBlank()) {
+            return false;
+        }
+        if (!MODE_CONSUME.equals(mode) && !MODE_SNAPSHOT.equals(mode)) {
+            throw new IllegalArgumentException(
+                    "network config 'mode' must be 'consume' or 'snapshot'");
+        }
+        return MODE_SNAPSHOT.equals(mode);
+    }
+
+    private static boolean parseBoolean(Object value, boolean fallback) {
+        String text = trimmed(value);
+        return text == null ? fallback : Boolean.parseBoolean(text);
+    }
+
+    private static String str(Object value) {
+        return value == null ? null : value.toString();
+    }
+
+    private static String trimmed(Object value) {
+        if (value == null) {
+            return null;
+        }
+        String text = value.toString().trim();
+        return text.isEmpty() ? null : text;
+    }
+
+    /** Never prints the credentials, so an accidental log line cannot leak them. */
+    @Override
+    public String toString() {
+        return "NetworkConfig[protocol="
+                + protocol
+                + ", host="
+                + host
+                + ", port="
+                + port
+                + ", username="
+                + username
+                + ", share="
+                + share
+                + ", directory="
+                + directory
+                + ", snapshot="
+                + snapshot
+                + ", recursive="
+                + recursive
+                + "]";
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/NetworkConnectionResolver.java

@@ -0,0 +1,136 @@
+package stirling.software.proprietary.policy.network;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.security.core.Authentication;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.security.core.userdetails.UserDetails;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import lombok.RequiredArgsConstructor;
+
+import stirling.software.proprietary.access.model.ResourceType;
+import stirling.software.proprietary.access.service.OwnershipService;
+import stirling.software.proprietary.integration.model.IntegrationConfig;
+import stirling.software.proprietary.integration.model.IntegrationType;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.security.model.User;
+import stirling.software.proprietary.security.service.UserService;
+
+import tools.jackson.core.type.TypeReference;
+import tools.jackson.databind.ObjectMapper;
+
+/**
+ * Turns a network source's options into a full {@link NetworkConfig} by dereferencing its {@code
+ * connectionId} to a stored NETWORK {@link IntegrationConfig} (the connection owns protocol, host,
+ * credentials and, for SMB, the share; the options own the per-use directory and mode). Options
+ * with no {@code connectionId} are parsed directly, so a programmatic caller can embed config.
+ *
+ * <p>Mirrors {@code S3ConnectionResolver}: an authenticated caller (save-time validation) must be
+ * allowed to use the connection; a background sweep with no caller skips that check, since the
+ * referencing source was access-checked when it was saved.
+ */
+@Service
+@RequiredArgsConstructor
+@Transactional(readOnly = true)
+public class NetworkConnectionResolver {
+
+    static final String CONNECTION_ID_OPTION = "connectionId";
+    private static final String DIRECTORY_OPTION = "directory";
+    private static final String MODE_OPTION = "mode";
+    private static final String RECURSIVE_OPTION = "recursive";
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private final IntegrationConfigRepository connections;
+    private final OwnershipService ownership;
+    private final UserService userService;
+
+    public NetworkConfig resolve(Map<String, Object> options) {
+        Long connectionId = connectionId(options);
+        if (connectionId == null) {
+            return NetworkConfig.from(options);
+        }
+        IntegrationConfig connection =
+                connections
+                        .findById(connectionId)
+                        .filter(cfg -> cfg.getIntegrationType() == IntegrationType.NETWORK)
+                        .filter(this::usableByCurrentUser)
+                        // Existence and access collapse into one error so a caller cannot tell
+                        // "no such connection" from "someone else's" and enumerate ids.
+                        .orElseThrow(
+                                () ->
+                                        new IllegalArgumentException(
+                                                "unknown or inaccessible network connection"));
+        if (!connection.isEnabled()) {
+            throw new IllegalArgumentException("network connection is disabled");
+        }
+        Map<String, Object> merged = new LinkedHashMap<>(connectionConfig(connection));
+        copyPerUseOption(options, merged, DIRECTORY_OPTION);
+        copyPerUseOption(options, merged, MODE_OPTION);
+        copyPerUseOption(options, merged, RECURSIVE_OPTION);
+        return NetworkConfig.from(merged);
+    }
+
+    static Long connectionId(Map<String, Object> options) {
+        Object reference = options.get(CONNECTION_ID_OPTION);
+        if (reference == null || (reference instanceof String s && s.isBlank())) {
+            return null;
+        }
+        if (reference instanceof Number number) {
+            return number.longValue();
+        }
+        try {
+            return Long.valueOf(reference.toString().trim());
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException(
+                    "network 'connectionId' is not a valid connection reference: " + reference);
+        }
+    }
+
+    private boolean usableByCurrentUser(IntegrationConfig connection) {
+        User user = currentUser();
+        return user == null || ownership.canUse(ResourceType.INTEGRATION_CONFIG, connection, user);
+    }
+
+    private User currentUser() {
+        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
+        if (auth == null || !auth.isAuthenticated()) {
+            return null;
+        }
+        Object principal = auth.getPrincipal();
+        if (principal instanceof User user) {
+            return user;
+        }
+        if (principal instanceof UserDetails userDetails) {
+            return userService.findByUsername(userDetails.getUsername()).orElse(null);
+        }
+        if (principal instanceof String username && !"anonymousUser".equals(username)) {
+            return userService.findByUsername(username).orElse(null);
+        }
+        return null;
+    }
+
+    private static Map<String, Object> connectionConfig(IntegrationConfig connection) {
+        String json = connection.getConfig();
+        if (json == null || json.isBlank()) {
+            return Map.of();
+        }
+        try {
+            return OBJECT_MAPPER.readValue(
+                    json, new TypeReference<LinkedHashMap<String, Object>>() {});
+        } catch (Exception e) {
+            throw new IllegalArgumentException(
+                    "network connection '" + connection.getName() + "' has unreadable config", e);
+        }
+    }
+
+    private static void copyPerUseOption(
+            Map<String, Object> options, Map<String, Object> merged, String key) {
+        Object value = options.get(key);
+        if (value != null && !value.toString().isBlank()) {
+            merged.put(key, value);
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/NetworkHostGuard.java

@@ -0,0 +1,68 @@
+package stirling.software.proprietary.policy.network;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+
+import stirling.software.common.model.ApplicationProperties;
+
+/**
+ * Refuses a network source whose host resolves to a loopback, link-local, or private address,
+ * unless the operator opts in via {@code policies.allowPrivateNetworkSources} (every host) or names
+ * the host in {@code policies.allowedPrivateNetworkHosts} (that host only, for shared infra). The
+ * host comes from a portal user, so without this a connection could be aimed at internal services
+ * (the cloud metadata address, an admin panel). Mirrors the S3 endpoint guard; enforced both at
+ * save time and before every connect.
+ */
+@Component
+@RequiredArgsConstructor
+public class NetworkHostGuard {
+
+    private final ApplicationProperties applicationProperties;
+
+    public void requirePermitted(String host) {
+        if (applicationProperties.getPolicies().isAllowPrivateNetworkSources()) {
+            return;
+        }
+        if (host == null || host.isBlank()) {
+            throw new IllegalArgumentException("network source requires a host");
+        }
+        if (isAllowlisted(host)) {
+            return;
+        }
+        InetAddress[] addresses;
+        try {
+            addresses = InetAddress.getAllByName(host);
+        } catch (UnknownHostException e) {
+            throw new IllegalArgumentException("cannot resolve network host '" + host + "'", e);
+        }
+        for (InetAddress address : addresses) {
+            if (isPrivateOrLocal(address)) {
+                throw new IllegalArgumentException(
+                        "network host '"
+                                + host
+                                + "' resolves to a private or local address ("
+                                + address.getHostAddress()
+                                + "); add it to policies.allowedPrivateNetworkHosts or set"
+                                + " policies.allowPrivateNetworkSources=true to allow an"
+                                + " on-network server");
+            }
+        }
+    }
+
+    private boolean isAllowlisted(String host) {
+        return applicationProperties.getPolicies().getAllowedPrivateNetworkHosts().stream()
+                .anyMatch(allowed -> allowed != null && allowed.trim().equalsIgnoreCase(host));
+    }
+
+    private static boolean isPrivateOrLocal(InetAddress address) {
+        return address.isLoopbackAddress()
+                || address.isLinkLocalAddress()
+                || address.isSiteLocalAddress()
+                || address.isAnyLocalAddress()
+                || address.isMulticastAddress();
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/NetworkIdentities.java

@@ -0,0 +1,52 @@
+package stirling.software.proprietary.policy.network;
+
+/**
+ * The network backend's identity and version scheme, shared by {@link NetworkInputSource} and its
+ * completion hook. Identity keys a file's ledger row and stays stable across sweeps; the gate is
+ * size + last-modified, since SFTP/FTP/SMB expose no ETag - the same "stat" strength as a folder
+ * source, so a touch that changes mtime counts as a new version.
+ */
+public final class NetworkIdentities {
+
+    private NetworkIdentities() {}
+
+    /** A stable, unique key for a remote file: {@code protocol://host:port[/share]/path}. */
+    public static String identity(NetworkConfig config, String path) {
+        StringBuilder id =
+                new StringBuilder(config.protocol().name().toLowerCase())
+                        .append("://")
+                        .append(config.host())
+                        .append(':')
+                        .append(config.port());
+        if (config.share() != null && !config.share().isBlank()) {
+            id.append('/').append(trimSlashes(config.share()));
+        }
+        id.append('/').append(trimLeadingSlash(path));
+        return id.toString();
+    }
+
+    /** The version gate: size and last-modified, mirroring the folder source's stat identity. */
+    public static String gate(long size, long lastModifiedMs) {
+        return size + ":" + lastModifiedMs;
+    }
+
+    private static String trimSlashes(String value) {
+        int start = 0;
+        int end = value.length();
+        while (start < end && (value.charAt(start) == '/' || value.charAt(start) == '\\')) {
+            start++;
+        }
+        while (end > start && (value.charAt(end - 1) == '/' || value.charAt(end - 1) == '\\')) {
+            end--;
+        }
+        return value.substring(start, end);
+    }
+
+    private static String trimLeadingSlash(String value) {
+        int start = 0;
+        while (start < value.length() && value.charAt(start) == '/') {
+            start++;
+        }
+        return value.substring(start);
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/NetworkInputSource.java

@@ -0,0 +1,148 @@
+package stirling.software.proprietary.policy.network;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.core.io.Resource;
+import org.springframework.stereotype.Service;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import stirling.software.proprietary.policy.input.InputSource;
+import stirling.software.proprietary.policy.input.ResolveContext;
+import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.model.PolicyInputs;
+
+/**
+ * Reads input files from a network file server (SFTP, FTP/FTPS, or SMB), one bean serving all three
+ * source types ({@code sftp}, {@code ftp}, {@code network}); the protocol comes from the referenced
+ * connection. Each listed file is its own unit of work, claimed through the {@link ResolveContext}
+ * ledger and tracked in place, mirroring the S3 source. The version gate is size + last-modified
+ * (there is no ETag), so this has folder-source "stat" strength. Options: "connectionId" references
+ * the stored NETWORK connection; "directory" is the poll folder (relative to the login home or
+ * share root); "recursive" descends into subdirectories; "mode" is "consume" (default: a processed
+ * file is deleted once every policy that claimed it has settled successfully and it is still the
+ * version that ran) or "snapshot" (stateless, every run sees the full set).
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class NetworkInputSource implements InputSource {
+
+    private final NetworkConnectionResolver connectionResolver;
+    private final RemoteFileClientFactory clientFactory;
+
+    @Override
+    public String type() {
+        return NetworkProtocol.SMB.sourceType();
+    }
+
+    @Override
+    public boolean supports(InputSpec spec) {
+        return spec != null && NetworkProtocol.forSourceType(spec.type()) != null;
+    }
+
+    /**
+     * Fails fast at save time: an unknown/disabled/inaccessible connection, a protocol that does
+     * not match the source type, a blocked host, or a server the connection cannot list.
+     */
+    @Override
+    public void validate(InputSpec spec) {
+        NetworkConfig config = connectionResolver.resolve(spec.options());
+        NetworkProtocol expected = NetworkProtocol.forSourceType(spec.type());
+        if (expected != null && config.protocol() != expected) {
+            throw new IllegalArgumentException(
+                    "source type '"
+                            + spec.type()
+                            + "' does not match the connection protocol "
+                            + config.protocol());
+        }
+        try (RemoteFileClient client = clientFactory.connect(config)) {
+            client.list(config.directory(), false);
+        } catch (IOException e) {
+            throw new IllegalArgumentException(
+                    "cannot access " + config.protocol() + " source: " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public List<ResolvedInput> resolve(InputSpec spec, ResolveContext ctx) throws IOException {
+        NetworkConfig config = connectionResolver.resolve(spec.options());
+        // A listing failure propagates so the sweep reads it as "could not list" (which vetoes
+        // presence cleanup), never as "verifiably no files".
+        List<RemoteFile> files;
+        try (RemoteFileClient client = clientFactory.connect(config)) {
+            files = client.list(config.directory(), config.recursive());
+        }
+        if (files.size() >= RemoteFileClient.MAX_FILES) {
+            log.warn(
+                    "Network source listing hit the {}-file cap; remaining files are picked up"
+                            + " on later sweeps",
+                    RemoteFileClient.MAX_FILES);
+        }
+
+        if (config.snapshot()) {
+            return files.stream()
+                    .map(file -> ResolvedInput.of(PolicyInputs.of(List.of(resource(config, file)))))
+                    .toList();
+        }
+
+        ctx.reportPresent(
+                files.stream()
+                        .map(file -> NetworkIdentities.identity(config, file.path()))
+                        .toList());
+
+        List<ResolvedInput> work = new ArrayList<>();
+        for (RemoteFile file : files) {
+            String identity = NetworkIdentities.identity(config, file.path());
+            String gate = NetworkIdentities.gate(file.size(), file.lastModifiedMs());
+            if (!ctx.claim(identity, gate, null)) {
+                continue;
+            }
+            work.add(
+                    new ResolvedInput(
+                            PolicyInputs.of(List.of(resource(config, file))),
+                            success ->
+                                    completeConsumed(ctx, config, file, identity, gate, success)));
+        }
+        return work;
+    }
+
+    /**
+     * Settle at the version this run claimed, then remove the file only when it still carries that
+     * version and every policy that claimed it has settled DONE, mirroring the S3 source's
+     * consensus delete. A failed run settles ERROR and never deletes; the DONE row of a file that
+     * could not be deleted still stops reprocessing.
+     */
+    private void completeConsumed(
+            ResolveContext ctx,
+            NetworkConfig config,
+            RemoteFile file,
+            String identity,
+            String claimGate,
+            boolean success) {
+        ctx.settle(identity, claimGate, null, success);
+        if (!success) {
+            return;
+        }
+        try (RemoteFileClient client = clientFactory.connect(config)) {
+            RemoteFile current = client.stat(file.path());
+            if (current == null) {
+                return; // removed by the user or a co-watching policy's consensus delete
+            }
+            String currentGate = NetworkIdentities.gate(current.size(), current.lastModifiedMs());
+            if (currentGate.equals(claimGate) && ctx.allSettledDone(identity)) {
+                client.delete(file.path());
+            }
+        } catch (IOException e) {
+            log.warn("Could not remove consumed network file {}: {}", identity, e.getMessage());
+        }
+    }
+
+    private Resource resource(NetworkConfig config, RemoteFile file) {
+        return new RemoteFileResource(clientFactory, config, file);
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/NetworkIntegrationValidator.java

@@ -0,0 +1,35 @@
+package stirling.software.proprietary.policy.network;
+
+import java.util.Map;
+
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+
+import stirling.software.proprietary.integration.model.IntegrationType;
+import stirling.software.proprietary.integration.service.IntegrationConfigValidator;
+
+/**
+ * The network connection schema, enforced when a NETWORK {@link IntegrationType} config is saved:
+ * protocol, host, credentials (and, for SMB, a share) required, with the same private-address guard
+ * {@link RemoteFileClientFactory} applies before connecting, moved to save time so a bad connection
+ * fails in the form rather than in a sweep. A live connectivity check runs later, when the source
+ * that uses the connection is saved ({@link NetworkInputSource#validate}).
+ */
+@Component
+@RequiredArgsConstructor
+public class NetworkIntegrationValidator implements IntegrationConfigValidator {
+
+    private final NetworkHostGuard hostGuard;
+
+    @Override
+    public IntegrationType type() {
+        return IntegrationType.NETWORK;
+    }
+
+    @Override
+    public void validate(Map<String, Object> config) {
+        NetworkConfig parsed = NetworkConfig.from(config);
+        hostGuard.requirePermitted(parsed.host());
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/NetworkProtocol.java

@@ -0,0 +1,57 @@
+package stirling.software.proprietary.policy.network;
+
+/**
+ * The file-transfer protocols a network source speaks. Each maps to a {@link RemoteFileClient}
+ * implementation and to a portal source type: {@code sftp} -> SFTP, {@code ftp} -> FTP/FTPS, {@code
+ * network} -> SMB (a Windows/Samba share). The stored connection carries the protocol; the source
+ * type only routes the UI.
+ */
+public enum NetworkProtocol {
+    SFTP(22, "sftp"),
+    FTP(21, "ftp"),
+    SMB(445, "network");
+
+    private final int defaultPort;
+    private final String sourceType;
+
+    NetworkProtocol(int defaultPort, String sourceType) {
+        this.defaultPort = defaultPort;
+        this.sourceType = sourceType;
+    }
+
+    public int defaultPort() {
+        return defaultPort;
+    }
+
+    /** The portal source {@code type} string this protocol backs. */
+    public String sourceType() {
+        return sourceType;
+    }
+
+    /** Parse a stored {@code protocol} option (case-insensitive), or null when absent/unknown. */
+    public static NetworkProtocol fromOption(String value) {
+        if (value == null) {
+            return null;
+        }
+        String normalized = value.trim().toUpperCase();
+        for (NetworkProtocol protocol : values()) {
+            if (protocol.name().equals(normalized)) {
+                return protocol;
+            }
+        }
+        return null;
+    }
+
+    /** The protocol a source {@code type} string implies, or null when it is not a network type. */
+    public static NetworkProtocol forSourceType(String type) {
+        if (type == null) {
+            return null;
+        }
+        for (NetworkProtocol protocol : values()) {
+            if (protocol.sourceType.equals(type)) {
+                return protocol;
+            }
+        }
+        return null;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/RemoteFile.java

@@ -0,0 +1,18 @@
+package stirling.software.proprietary.policy.network;
+
+/**
+ * One remote file discovered by a {@link RemoteFileClient} listing: its full path on the server,
+ * its display name, size in bytes, and last-modified epoch milliseconds. Size and mtime form the
+ * version gate ({@link NetworkIdentities#gate}) since network protocols expose no ETag.
+ */
+public record RemoteFile(String path, String name, long size, long lastModifiedMs) {
+
+    public RemoteFile {
+        if (path == null || path.isBlank()) {
+            throw new IllegalArgumentException("remote file requires a path");
+        }
+        if (name == null || name.isBlank()) {
+            name = path.substring(path.lastIndexOf('/') + 1);
+        }
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/RemoteFileClient.java

@@ -0,0 +1,38 @@
+package stirling.software.proprietary.policy.network;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+
+/**
+ * A connected session to one network file server (SFTP, FTP, or SMB), created by {@link
+ * RemoteFileClientFactory}. Sessions are short-lived and single-use: the caller opens one, performs
+ * a listing, a read, or a delete, then closes it - there is no pooling, because network sessions
+ * time out and go stale far more readily than an S3 HTTP client. Callers must {@link #close()}
+ * every client (try-with-resources), except when streaming, where ownership is handed to the
+ * returned stream so it can be read after {@code resolve} returns.
+ */
+public interface RemoteFileClient extends Closeable {
+
+    /**
+     * Listing cap: a listing stops once this many files are collected, bounding memory against a
+     * huge (or maliciously deep) tree. Consume mode drains the rest on later sweeps.
+     */
+    int MAX_FILES = 10_000;
+
+    /**
+     * Every regular, non-hidden file under {@code directory}, optionally descending into it,
+     * truncated at {@link #MAX_FILES}.
+     */
+    List<RemoteFile> list(String directory, boolean recursive) throws IOException;
+
+    /** The current metadata for one path, or null when it no longer exists. */
+    RemoteFile stat(String path) throws IOException;
+
+    /** Opens the file for reading; the stream stays valid until this client is closed. */
+    InputStream open(String path) throws IOException;
+
+    /** Removes the file; a no-op if it is already gone. */
+    void delete(String path) throws IOException;
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/RemoteFileClientFactory.java

@@ -0,0 +1,28 @@
+package stirling.software.proprietary.policy.network;
+
+import java.io.IOException;
+
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Opens a fresh {@link RemoteFileClient} for one {@link NetworkConfig}, dispatched by protocol. The
+ * host is guarded against private addresses before every connect, since it comes from a portal user
+ * and each operation opens its own short-lived session (there is no long-lived pool to guard once).
+ */
+@Component
+@RequiredArgsConstructor
+public class RemoteFileClientFactory {
+
+    private final NetworkHostGuard hostGuard;
+
+    public RemoteFileClient connect(NetworkConfig config) throws IOException {
+        hostGuard.requirePermitted(config.host());
+        return switch (config.protocol()) {
+            case SFTP -> SftpFileClient.connect(config);
+            case FTP -> FtpFileClient.connect(config);
+            case SMB -> SmbFileClient.connect(config);
+        };
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/RemoteFileResource.java

@@ -0,0 +1,68 @@
+package stirling.software.proprietary.policy.network;
+
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.springframework.core.io.AbstractResource;
+
+/**
+ * A policy input backed by a remote file, streamed on demand. Each read opens its own short-lived
+ * client (the listing client is long gone by the time the run reads the file) and hands it to the
+ * stream, so closing the stream tears the session down. Size and name come from the listing, so the
+ * pipeline can size the input without a second round trip.
+ */
+final class RemoteFileResource extends AbstractResource {
+
+    private final RemoteFileClientFactory factory;
+    private final NetworkConfig config;
+    private final RemoteFile file;
+
+    RemoteFileResource(RemoteFileClientFactory factory, NetworkConfig config, RemoteFile file) {
+        this.factory = factory;
+        this.config = config;
+        this.file = file;
+    }
+
+    @Override
+    public InputStream getInputStream() throws IOException {
+        RemoteFileClient client = factory.connect(config);
+        try {
+            InputStream stream = client.open(file.path());
+            return new FilterInputStream(stream) {
+                @Override
+                public void close() throws IOException {
+                    try {
+                        super.close();
+                    } finally {
+                        client.close();
+                    }
+                }
+            };
+        } catch (IOException e) {
+            client.close();
+            throw e;
+        }
+    }
+
+    /** Listed just now; a reader gets a precise error from {@link #getInputStream} instead. */
+    @Override
+    public boolean exists() {
+        return true;
+    }
+
+    @Override
+    public long contentLength() {
+        return file.size();
+    }
+
+    @Override
+    public String getFilename() {
+        return file.name();
+    }
+
+    @Override
+    public String getDescription() {
+        return "network file " + NetworkIdentities.identity(config, file.path());
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/SftpFileClient.java

@@ -0,0 +1,312 @@
+package stirling.software.proprietary.policy.network;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.List;
+import java.util.Locale;
+import java.util.Vector;
+
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.HostKey;
+import com.jcraft.jsch.HostKeyRepository;
+import com.jcraft.jsch.JSch;
+import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.Session;
+import com.jcraft.jsch.SftpATTRS;
+import com.jcraft.jsch.SftpException;
+import com.jcraft.jsch.UserInfo;
+
+import stirling.software.common.configuration.InstallationPathConfig;
+
+/**
+ * SFTP client over jsch. Host keys are verified: a connection with a configured {@code
+ * hostKeyFingerprint} only accepts the server presenting that key; otherwise the key seen on the
+ * first connect is pinned in a known-hosts file under the config directory and any later change is
+ * refused (trust-on-first-use). Directories are listed, read and deleted through one channel; a
+ * blank directory means the login home. Symlinked directories are not followed and hidden entries
+ * are skipped, mirroring the folder source.
+ */
+final class SftpFileClient implements RemoteFileClient {
+
+    private static final int CONNECT_TIMEOUT_MS = 15_000;
+    private static final int READ_TIMEOUT_MS = 60_000;
+    private static final int MAX_DEPTH = 64;
+    static final String KNOWN_HOSTS_FILE = "sftp_known_hosts";
+
+    private final Session session;
+    private final ChannelSftp channel;
+
+    private SftpFileClient(Session session, ChannelSftp channel) {
+        this.session = session;
+        this.channel = channel;
+    }
+
+    static SftpFileClient connect(NetworkConfig config) throws IOException {
+        JSch jsch = new JSch();
+        try {
+            if (config.privateKey() != null) {
+                byte[] passphrase =
+                        config.privateKeyPassphrase() == null
+                                ? null
+                                : config.privateKeyPassphrase().getBytes(StandardCharsets.UTF_8);
+                jsch.addIdentity(
+                        "network-source",
+                        config.privateKey().getBytes(StandardCharsets.UTF_8),
+                        null,
+                        passphrase);
+            }
+            Session session = jsch.getSession(config.username(), config.host(), config.port());
+            if (config.password() != null) {
+                session.setPassword(config.password());
+            }
+            if (config.hostKeyFingerprint() != null) {
+                // Pinned key: only the configured fingerprint is ever accepted.
+                session.setHostKeyRepository(
+                        new PinnedHostKeyRepository(config.hostKeyFingerprint()));
+            } else {
+                // Trust-on-first-use: the first key seen is recorded, a changed key is refused.
+                jsch.setKnownHosts(knownHostsFile().toString());
+                session.setHostKeyRepository(
+                        new TofuHostKeyRepository(jsch.getHostKeyRepository()));
+            }
+            session.setConfig("StrictHostKeyChecking", "yes");
+            session.connect(CONNECT_TIMEOUT_MS);
+            session.setTimeout(READ_TIMEOUT_MS);
+            ChannelSftp channel = (ChannelSftp) session.openChannel("sftp");
+            channel.connect(CONNECT_TIMEOUT_MS);
+            return new SftpFileClient(session, channel);
+        } catch (JSchException e) {
+            throw new IOException(
+                    "SFTP connection to " + config.host() + " failed: " + e.getMessage(), e);
+        }
+    }
+
+    /** jsch only persists newly pinned keys into a file that already exists, so create it. */
+    private static Path knownHostsFile() throws IOException {
+        Path file = Path.of(InstallationPathConfig.getConfigPath(), KNOWN_HOSTS_FILE);
+        Files.createDirectories(file.getParent());
+        try {
+            Files.createFile(file);
+        } catch (FileAlreadyExistsException ignored) {
+            // Another connect already created it; the content is managed by jsch.
+        }
+        return file;
+    }
+
+    /**
+     * Trust-on-first-use over the shared known-hosts file: an unknown host's key is pinned on first
+     * contact, and a key that later differs is refused ({@code CHANGED}). Rejection happens before
+     * authentication, so credentials are never sent to an impersonating server.
+     */
+    private static final class TofuHostKeyRepository implements HostKeyRepository {
+
+        private final HostKeyRepository knownHosts;
+
+        private TofuHostKeyRepository(HostKeyRepository knownHosts) {
+            this.knownHosts = knownHosts;
+        }
+
+        @Override
+        public int check(String host, byte[] key) {
+            int result = knownHosts.check(host, key);
+            if (result != NOT_INCLUDED) {
+                return result;
+            }
+            try {
+                knownHosts.add(new HostKey(host, key), null);
+            } catch (JSchException e) {
+                return CHANGED; // a key we cannot even parse is refused, not trusted
+            }
+            return OK;
+        }
+
+        @Override
+        public void add(HostKey hostkey, UserInfo ui) {
+            knownHosts.add(hostkey, ui);
+        }
+
+        @Override
+        public void remove(String host, String type) {
+            knownHosts.remove(host, type);
+        }
+
+        @Override
+        public void remove(String host, String type, byte[] key) {
+            knownHosts.remove(host, type, key);
+        }
+
+        @Override
+        public String getKnownHostsRepositoryID() {
+            return knownHosts.getKnownHostsRepositoryID();
+        }
+
+        @Override
+        public HostKey[] getHostKey() {
+            return knownHosts.getHostKey();
+        }
+
+        @Override
+        public HostKey[] getHostKey(String host, String type) {
+            return knownHosts.getHostKey(host, type);
+        }
+    }
+
+    /**
+     * Accepts only the server key whose SHA-256 fingerprint matches the configured value (OpenSSH
+     * {@code SHA256:base64} form, prefix optional). Rejection happens before authentication, so
+     * credentials are never sent to an impersonating server.
+     */
+    private static final class PinnedHostKeyRepository implements HostKeyRepository {
+
+        private final String expected;
+
+        private PinnedHostKeyRepository(String fingerprint) {
+            this.expected = normalize(fingerprint);
+        }
+
+        @Override
+        public int check(String host, byte[] key) {
+            return expected.equals(sha256Fingerprint(key)) ? OK : CHANGED;
+        }
+
+        private static String normalize(String fingerprint) {
+            String value = fingerprint.trim();
+            if (value.toUpperCase(Locale.ROOT).startsWith("SHA256:")) {
+                value = value.substring("SHA256:".length());
+            }
+            // OpenSSH prints the digest base64 without padding; accept it padded too.
+            return value.replace("=", "");
+        }
+
+        private static String sha256Fingerprint(byte[] key) {
+            try {
+                MessageDigest digest = MessageDigest.getInstance("SHA-256");
+                return Base64.getEncoder().withoutPadding().encodeToString(digest.digest(key));
+            } catch (NoSuchAlgorithmException e) {
+                throw new IllegalStateException("SHA-256 unavailable", e);
+            }
+        }
+
+        @Override
+        public void add(HostKey hostkey, UserInfo ui) {}
+
+        @Override
+        public void remove(String host, String type) {}
+
+        @Override
+        public void remove(String host, String type, byte[] key) {}
+
+        @Override
+        public String getKnownHostsRepositoryID() {
+            return "pinned-fingerprint";
+        }
+
+        @Override
+        public HostKey[] getHostKey() {
+            return new HostKey[0];
+        }
+
+        @Override
+        public HostKey[] getHostKey(String host, String type) {
+            return new HostKey[0];
+        }
+    }
+
+    @Override
+    public List<RemoteFile> list(String directory, boolean recursive) throws IOException {
+        List<RemoteFile> files = new ArrayList<>();
+        collect(dir(directory), recursive, 0, files);
+        return files;
+    }
+
+    @SuppressWarnings("unchecked")
+    private void collect(String directory, boolean recursive, int depth, List<RemoteFile> out)
+            throws IOException {
+        Vector<ChannelSftp.LsEntry> entries;
+        try {
+            entries = channel.ls(directory);
+        } catch (SftpException e) {
+            throw new IOException("cannot list " + directory + ": " + e.getMessage(), e);
+        }
+        for (ChannelSftp.LsEntry entry : entries) {
+            if (out.size() >= MAX_FILES) {
+                return;
+            }
+            String name = entry.getFilename();
+            if (name.equals(".") || name.equals("..") || name.startsWith(".")) {
+                continue;
+            }
+            SftpATTRS attrs = entry.getAttrs();
+            String path = join(directory, name);
+            if (attrs.isDir()) {
+                if (recursive && !attrs.isLink() && depth < MAX_DEPTH) {
+                    collect(path, true, depth + 1, out);
+                }
+                continue;
+            }
+            if (attrs.isReg()) {
+                out.add(new RemoteFile(path, name, attrs.getSize(), attrs.getMTime() * 1000L));
+            }
+        }
+    }
+
+    @Override
+    public RemoteFile stat(String path) throws IOException {
+        try {
+            SftpATTRS attrs = channel.stat(path);
+            String name = path.substring(path.lastIndexOf('/') + 1);
+            return new RemoteFile(path, name, attrs.getSize(), attrs.getMTime() * 1000L);
+        } catch (SftpException e) {
+            if (e.id == ChannelSftp.SSH_FX_NO_SUCH_FILE) {
+                return null;
+            }
+            throw new IOException("cannot stat " + path + ": " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public InputStream open(String path) throws IOException {
+        try {
+            return channel.get(path);
+        } catch (SftpException e) {
+            throw new IOException("cannot read " + path + ": " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void delete(String path) throws IOException {
+        try {
+            channel.rm(path);
+        } catch (SftpException e) {
+            if (e.id == ChannelSftp.SSH_FX_NO_SUCH_FILE) {
+                return;
+            }
+            throw new IOException("cannot delete " + path + ": " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void close() {
+        channel.disconnect();
+        session.disconnect();
+    }
+
+    private static String dir(String directory) {
+        return directory == null || directory.isBlank() ? "." : directory;
+    }
+
+    private static String join(String directory, String name) {
+        if (directory.equals(".")) {
+            return name;
+        }
+        return directory.endsWith("/") ? directory + name : directory + "/" + name;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/network/SmbFileClient.java

@@ -0,0 +1,236 @@
+package stirling.software.proprietary.policy.network;
+
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import com.hierynomus.msdtyp.AccessMask;
+import com.hierynomus.mserref.NtStatus;
+import com.hierynomus.msfscc.FileAttributes;
+import com.hierynomus.msfscc.fileinformation.FileAllInformation;
+import com.hierynomus.msfscc.fileinformation.FileIdBothDirectoryInformation;
+import com.hierynomus.mssmb2.SMB2CreateDisposition;
+import com.hierynomus.mssmb2.SMB2ShareAccess;
+import com.hierynomus.mssmb2.SMBApiException;
+import com.hierynomus.protocol.commons.socket.ProxySocketFactory;
+import com.hierynomus.smbj.SMBClient;
+import com.hierynomus.smbj.SmbConfig;
+import com.hierynomus.smbj.auth.AuthenticationContext;
+import com.hierynomus.smbj.common.SMBRuntimeException;
+import com.hierynomus.smbj.connection.Connection;
+import com.hierynomus.smbj.session.Session;
+import com.hierynomus.smbj.share.DiskShare;
+
+/**
+ * SMB2/3 client over smbj against a Windows or Samba share. The share is named on the connection;
+ * directories are relative to its root, using backslashes as SMB expects. A blank directory lists
+ * the share root. Hidden and system entries are skipped, and symlink/reparse dirs are not
+ * descended, mirroring the folder source.
+ */
+final class SmbFileClient implements RemoteFileClient {
+
+    private static final int CONNECT_TIMEOUT_MS = 15_000;
+    private static final int READ_TIMEOUT_MS = 60_000;
+    private static final int MAX_DEPTH = 64;
+
+    private final SMBClient smbClient;
+    private final Connection connection;
+    private final Session session;
+    private final DiskShare share;
+
+    private SmbFileClient(
+            SMBClient smbClient, Connection connection, Session session, DiskShare share) {
+        this.smbClient = smbClient;
+        this.connection = connection;
+        this.session = session;
+        this.share = share;
+    }
+
+    static SmbFileClient connect(NetworkConfig config) throws IOException {
+        // Bounded connect and read timeouts (both default to unlimited) so a host that accepts
+        // the TCP connection then stalls cannot wedge a poller.
+        SmbConfig smbConfig =
+                SmbConfig.builder()
+                        .withSocketFactory(new ProxySocketFactory(CONNECT_TIMEOUT_MS))
+                        .withSoTimeout(READ_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+                        .withTimeout(READ_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+                        .build();
+        SMBClient smbClient = new SMBClient(smbConfig);
+        Connection connection = smbClient.connect(config.host(), config.port());
+        try {
+            char[] password =
+                    config.password() == null ? new char[0] : config.password().toCharArray();
+            AuthenticationContext auth =
+                    new AuthenticationContext(config.username(), password, config.domain());
+            Session session = connection.authenticate(auth);
+            DiskShare share = (DiskShare) session.connectShare(config.share());
+            return new SmbFileClient(smbClient, connection, session, share);
+        } catch (RuntimeException e) {
+            closeQuietly(connection);
+            closeQuietly(smbClient);
+            throw new IOException(
+                    "SMB connection to \\\\"
+                            + config.host()
+                            + "\\"
+                            + config.share()
+                            + " failed: "
+                            + e.getMessage(),
+                    e);
+        }
+    }
+
+    @Override
+    public List<RemoteFile> list(String directory, boolean recursive) throws IOException {
+        List<RemoteFile> files = new ArrayList<>();
+        collect(smbDir(directory), recursive, 0, files);
+        return files;
+    }
+
+    private void collect(String directory, boolean recursive, int depth, List<RemoteFile> out)
+            throws IOException {
+        List<FileIdBothDirectoryInformation> entries;
+        try {
+            entries = share.list(directory);
+        } catch (SMBRuntimeException e) {
+            // smbj throws unchecked; surface as IOException like the SFTP/FTP clients.
+            throw new IOException("cannot list " + slashed(directory) + ": " + e.getMessage(), e);
+        }
+        for (FileIdBothDirectoryInformation info : entries) {
+            if (out.size() >= MAX_FILES) {
+                return;
+            }
+            String name = info.getFileName();
+            if (name.equals(".") || name.equals("..") || name.startsWith(".")) {
+                continue;
+            }
+            long attributes = info.getFileAttributes();
+            if (isSet(attributes, FileAttributes.FILE_ATTRIBUTE_HIDDEN)
+                    || isSet(attributes, FileAttributes.FILE_ATTRIBUTE_SYSTEM)) {
+                continue;
+            }
+            String path = join(directory, name);
+            if (isSet(attributes, FileAttributes.FILE_ATTRIBUTE_DIRECTORY)) {
+                if (recursive
+                        && !isSet(attributes, FileAttributes.FILE_ATTRIBUTE_REPARSE_POINT)
+                        && depth < MAX_DEPTH) {
+                    collect(path, true, depth + 1, out);
+                }
+                continue;
+            }
+            out.add(
+                    new RemoteFile(
+                            slashed(path),
+                            name,
+                            info.getEndOfFile(),
+                            info.getLastWriteTime().toEpochMillis()));
+        }
+    }
+
+    @Override
+    public RemoteFile stat(String path) throws IOException {
+        String smbPath = smbPath(path);
+        try {
+            FileAllInformation info = share.getFileInformation(smbPath);
+            String name = path.substring(path.lastIndexOf('/') + 1);
+            return new RemoteFile(
+                    slashed(smbPath),
+                    name,
+                    info.getStandardInformation().getEndOfFile(),
+                    info.getBasicInformation().getLastWriteTime().toEpochMillis());
+        } catch (SMBApiException e) {
+            if (isNotFound(e)) {
+                return null;
+            }
+            throw new IOException("cannot stat " + path + ": " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public InputStream open(String path) throws IOException {
+        try {
+            com.hierynomus.smbj.share.File file =
+                    share.openFile(
+                            smbPath(path),
+                            EnumSet.of(AccessMask.GENERIC_READ),
+                            null,
+                            SMB2ShareAccess.ALL,
+                            SMB2CreateDisposition.FILE_OPEN,
+                            null);
+            // Close the SMB file handle when the stream is closed, before the session is torn down.
+            return new FilterInputStream(file.getInputStream()) {
+                @Override
+                public void close() throws IOException {
+                    try {
+                        super.close();
+                    } finally {
+                        file.close();
+                    }
+                }
+            };
+        } catch (SMBApiException e) {
+            throw new IOException("cannot read " + path + ": " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void delete(String path) throws IOException {
+        try {
+            share.rm(smbPath(path));
+        } catch (SMBApiException e) {
+            if (isNotFound(e)) {
+                return;
+            }
+            throw new IOException("cannot delete " + path + ": " + e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void close() {
+        closeQuietly(share);
+        closeQuietly(session);
+        closeQuietly(connection);
+        closeQuietly(smbClient);
+    }
+
+    private static boolean isNotFound(SMBApiException e) {
+        return e.getStatus() == NtStatus.STATUS_OBJECT_NAME_NOT_FOUND
+                || e.getStatus() == NtStatus.STATUS_OBJECT_PATH_NOT_FOUND;
+    }
+
+    private static boolean isSet(long attributes, FileAttributes attribute) {
+        return (attributes & attribute.getValue()) != 0;
+    }
+
+    private static void closeQuietly(AutoCloseable closeable) {
+        try {
+            closeable.close();
+        } catch (Exception ignored) {
+            // Best-effort teardown; the session is being abandoned anyway.
+        }
+    }
+
+    /** The share-relative listing path: blank means the share root, which smbj lists as "". */
+    private static String smbDir(String directory) {
+        return directory == null || directory.isBlank() ? "" : smbPath(directory);
+    }
+
+    /** SMB paths use backslashes; internal identities keep forward slashes. */
+    private static String smbPath(String path) {
+        return path.replace('/', '\\');
+    }
+
+    private static String slashed(String path) {
+        return path.replace('\\', '/');
+    }
+
+    private static String join(String directory, String name) {
+        if (directory.isEmpty()) {
+            return name;
+        }
+        return directory.endsWith("\\") ? directory + name : directory + "\\" + name;
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/network/FtpNetworkSourceIntegrationTest.java

@@ -0,0 +1,199 @@
+package stirling.software.proprietary.policy.network;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.FixedHostPortGenericContainer;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.access.service.OwnershipService;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.policy.input.ResolveContext;
+import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.security.service.UserService;
+
+/**
+ * End-to-end {@link NetworkInputSource} test over FTP against a real vsftpd server, through the
+ * production {@link FtpFileClient}: listing, claiming, streaming the actual bytes (exercising the
+ * passive data channel and {@code completePendingCommand}), consume delete, and save-time
+ * validation. Passive FTP hands the client a data port, so the control and one passive port are
+ * pinned to fixed host ports and the server advertises 127.0.0.1 - the only way this works through
+ * Testcontainers' per-run port mapping.
+ */
+@Testcontainers(disabledWithoutDocker = true)
+@EnabledIfEnvironmentVariable(
+        named = "RUN_NETWORK_INTEGRATION_TESTS",
+        matches = "true",
+        disabledReason =
+                "Spins up SFTP/FTP/SMB containers; opt-in to keep several heavy containers off the"
+                        + " standard CI runner. Run with RUN_NETWORK_INTEGRATION_TESTS=true.")
+class FtpNetworkSourceIntegrationTest {
+
+    private static final String POLICY = "p1";
+    private static final String USER = "stirling";
+    private static final String PASS = "secret";
+    private static final String HOME = "/ftp/stirling";
+    // Passive FTP needs the data port pinned to a known host port, so pick two free ones up front
+    // (rather than hard-coding) to avoid clashing with anything else on a CI runner.
+    private static final int CONTROL_PORT;
+    private static final int PASSIVE_PORT;
+
+    static {
+        try (java.net.ServerSocket control = new java.net.ServerSocket(0);
+                java.net.ServerSocket passive = new java.net.ServerSocket(0)) {
+            CONTROL_PORT = control.getLocalPort();
+            PASSIVE_PORT = passive.getLocalPort();
+        } catch (java.io.IOException e) {
+            throw new java.io.UncheckedIOException(e);
+        }
+    }
+
+    @org.testcontainers.junit.jupiter.Container
+    static GenericContainer<?> ftp =
+            new FixedHostPortGenericContainer<>("delfer/alpine-ftp-server:latest")
+                    .withFixedExposedPort(CONTROL_PORT, 21)
+                    .withFixedExposedPort(PASSIVE_PORT, PASSIVE_PORT)
+                    .withEnv("USERS", USER + "|" + PASS + "|" + HOME + "|1000")
+                    .withEnv("ADDRESS", "127.0.0.1")
+                    .withEnv("MIN_PORT", String.valueOf(PASSIVE_PORT))
+                    .withEnv("MAX_PORT", String.valueOf(PASSIVE_PORT))
+                    .waitingFor(Wait.forListeningPort());
+
+    private NetworkInputSource source;
+    private InProcessProcessedLedger ledger;
+    private RecordingContext ctx;
+
+    @BeforeEach
+    void setUp() {
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateNetworkSources(true);
+        RemoteFileClientFactory factory =
+                new RemoteFileClientFactory(new NetworkHostGuard(properties));
+        NetworkConnectionResolver resolver =
+                new NetworkConnectionResolver(
+                        mock(IntegrationConfigRepository.class),
+                        mock(OwnershipService.class),
+                        mock(UserService.class));
+        source = new NetworkInputSource(resolver, factory);
+        ledger = new InProcessProcessedLedger();
+        ctx = new RecordingContext();
+    }
+
+    @Test
+    void consumeListsStreamsAndDeletes() throws Exception {
+        put("doc.pdf", "hello ftp");
+
+        List<ResolvedInput> work = source.resolve(spec(Map.of()), ctx);
+
+        assertThat(work).hasSize(1);
+        assertThat(read(work.get(0))).isEqualTo("hello ftp");
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+
+        work.get(0).onComplete().accept(true);
+        assertThat(exists("doc.pdf")).isFalse();
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+    }
+
+    @Test
+    void aFailedFileStaysOnTheServer() throws Exception {
+        put("doc.pdf", "data");
+
+        source.resolve(spec(Map.of()), ctx).get(0).onComplete().accept(false);
+
+        assertThat(exists("doc.pdf")).isTrue();
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+    }
+
+    @Test
+    void validateConnectsAndRejectsBadCredentials() {
+        source.validate(spec(Map.of()));
+
+        Map<String, Object> wrong = new HashMap<>(baseOptions());
+        wrong.put("password", "not-the-password");
+        assertThatThrownBy(() -> source.validate(new InputSpec("ftp", wrong)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("cannot access");
+    }
+
+    private Map<String, Object> baseOptions() {
+        Map<String, Object> options = new HashMap<>();
+        options.put("protocol", "ftp");
+        options.put("host", "127.0.0.1");
+        options.put("port", String.valueOf(CONTROL_PORT));
+        options.put("username", USER);
+        options.put("password", PASS);
+        return options;
+    }
+
+    private InputSpec spec(Map<String, Object> extra) {
+        Map<String, Object> options = new HashMap<>(baseOptions());
+        options.putAll(extra);
+        return new InputSpec("ftp", options);
+    }
+
+    private void put(String name, String content) throws Exception {
+        String path = HOME + "/" + name;
+        exec("printf '%s' '" + content + "' > " + path + " && chown 1000 " + path);
+    }
+
+    private boolean exists(String name) throws Exception {
+        return exec("test -f " + HOME + "/" + name + " && echo yes || echo no").contains("yes");
+    }
+
+    private String exec(String script) throws Exception {
+        Container.ExecResult result = ftp.execInContainer("sh", "-c", script);
+        return result.getStdout() + result.getStderr();
+    }
+
+    private static String read(ResolvedInput unit) throws IOException {
+        try (InputStream stream = unit.inputs().primary().get(0).getInputStream()) {
+            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
+        }
+    }
+
+    private class RecordingContext implements ResolveContext {
+
+        private final List<String> present = new ArrayList<>();
+
+        @Override
+        public boolean claim(String identity, String gate, Supplier<String> contentHash) {
+            return ledger.claim(POLICY, identity, gate, contentHash);
+        }
+
+        @Override
+        public void settle(
+                String identity, String finalGate, String finalContentHash, boolean success) {
+            ledger.settle(POLICY, identity, finalGate, finalContentHash, success);
+        }
+
+        @Override
+        public boolean allSettledDone(String identity) {
+            return ledger.allSettledDone(identity);
+        }
+
+        @Override
+        public void reportPresent(Collection<String> identities) {
+            present.addAll(identities);
+        }
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/network/NetworkConfigTest.java

@@ -0,0 +1,136 @@
+package stirling.software.proprietary.policy.network;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+class NetworkConfigTest {
+
+    private static Map<String, Object> base(String protocol) {
+        Map<String, Object> options = new LinkedHashMap<>();
+        options.put("protocol", protocol);
+        options.put("host", "files.example.com");
+        options.put("username", "u");
+        options.put("password", "p");
+        return options;
+    }
+
+    @Test
+    void parsesAnSftpConnectionWithDefaultPort() {
+        NetworkConfig config = NetworkConfig.from(base("sftp"));
+        assertEquals(NetworkProtocol.SFTP, config.protocol());
+        assertEquals(22, config.port());
+        assertEquals("files.example.com", config.host());
+    }
+
+    @Test
+    void defaultsPortPerProtocol() {
+        assertEquals(21, NetworkConfig.from(base("ftp")).port());
+        Map<String, Object> smb = base("smb");
+        smb.put("share", "docs");
+        assertEquals(445, NetworkConfig.from(smb).port());
+    }
+
+    @Test
+    void implicitFtpsDefaultsToPort990() {
+        Map<String, Object> implicit = base("ftp");
+        implicit.put("security", "implicit");
+        assertEquals(990, NetworkConfig.from(implicit).port());
+
+        implicit.put("port", "2121");
+        assertEquals(2121, NetworkConfig.from(implicit).port());
+    }
+
+    @Test
+    void sftpAcceptsAPrivateKeyInsteadOfAPassword() {
+        Map<String, Object> options = base("sftp");
+        options.remove("password");
+        options.put("privateKey", "dummy-test-key-material");
+        NetworkConfig config = NetworkConfig.from(options);
+        assertEquals(NetworkProtocol.SFTP, config.protocol());
+    }
+
+    @Test
+    void sftpWithoutPasswordOrKeyIsRejected() {
+        Map<String, Object> options = base("sftp");
+        options.remove("password");
+        assertThrows(IllegalArgumentException.class, () -> NetworkConfig.from(options));
+    }
+
+    @Test
+    void ftpRequiresAPassword() {
+        Map<String, Object> options = base("ftp");
+        options.remove("password");
+        assertThrows(IllegalArgumentException.class, () -> NetworkConfig.from(options));
+    }
+
+    @Test
+    void smbRequiresAShare() {
+        assertThrows(IllegalArgumentException.class, () -> NetworkConfig.from(base("smb")));
+    }
+
+    @Test
+    void requiresProtocolHostAndUsername() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> NetworkConfig.from(Map.of("host", "h", "username", "u", "password", "p")));
+        Map<String, Object> noHost = base("sftp");
+        noHost.remove("host");
+        assertThrows(IllegalArgumentException.class, () -> NetworkConfig.from(noHost));
+        Map<String, Object> noUser = base("sftp");
+        noUser.remove("username");
+        assertThrows(IllegalArgumentException.class, () -> NetworkConfig.from(noUser));
+    }
+
+    @Test
+    void rejectsAnOutOfRangePort() {
+        Map<String, Object> options = base("sftp");
+        options.put("port", "70000");
+        assertThrows(IllegalArgumentException.class, () -> NetworkConfig.from(options));
+    }
+
+    @Test
+    void rejectsATraversalDirectory() {
+        Map<String, Object> options = base("sftp");
+        options.put("directory", "in/../../etc");
+        assertThrows(IllegalArgumentException.class, () -> NetworkConfig.from(options));
+    }
+
+    @Test
+    void mapsModeToSnapshotFlag() {
+        Map<String, Object> consume = base("sftp");
+        consume.put("mode", "consume");
+        assertFalse(NetworkConfig.from(consume).snapshot());
+
+        Map<String, Object> snapshot = base("sftp");
+        snapshot.put("mode", "snapshot");
+        assertTrue(NetworkConfig.from(snapshot).snapshot());
+
+        Map<String, Object> bad = base("sftp");
+        bad.put("mode", "sometimes");
+        assertThrows(IllegalArgumentException.class, () -> NetworkConfig.from(bad));
+    }
+
+    @Test
+    void carriesTheHostKeyFingerprint() {
+        Map<String, Object> options = base("sftp");
+        options.put("hostKeyFingerprint", " SHA256:abc123 ");
+        assertEquals("SHA256:abc123", NetworkConfig.from(options).hostKeyFingerprint());
+        assertNull(NetworkConfig.from(base("sftp")).hostKeyFingerprint());
+    }
+
+    @Test
+    void toStringHidesSecrets() {
+        Map<String, Object> options = base("sftp");
+        options.put("password", "hunter2");
+        String text = NetworkConfig.from(options).toString();
+        assertFalse(text.contains("hunter2"), "password must not appear in toString");
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/network/NetworkHostGuardTest.java

@@ -0,0 +1,64 @@
+package stirling.software.proprietary.policy.network;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.model.ApplicationProperties;
+
+class NetworkHostGuardTest {
+
+    private static NetworkHostGuard guard(boolean allowPrivate) {
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateNetworkSources(allowPrivate);
+        return new NetworkHostGuard(properties);
+    }
+
+    private static NetworkHostGuard guardWithAllowlist(String... hosts) {
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowedPrivateNetworkHosts(List.of(hosts));
+        return new NetworkHostGuard(properties);
+    }
+
+    @Test
+    void rejectsLoopbackByDefault() {
+        assertThrows(
+                IllegalArgumentException.class, () -> guard(false).requirePermitted("127.0.0.1"));
+    }
+
+    @Test
+    void rejectsPrivateRangesByDefault() {
+        assertThrows(
+                IllegalArgumentException.class, () -> guard(false).requirePermitted("10.0.0.1"));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> guard(false).requirePermitted("192.168.1.10"));
+    }
+
+    @Test
+    void rejectsABlankHost() {
+        assertThrows(IllegalArgumentException.class, () -> guard(false).requirePermitted(""));
+    }
+
+    @Test
+    void optInAllowsPrivateAddresses() {
+        assertDoesNotThrow(() -> guard(true).requirePermitted("127.0.0.1"));
+        assertDoesNotThrow(() -> guard(true).requirePermitted("10.0.0.1"));
+    }
+
+    @Test
+    void allowlistedHostIsAllowedWithoutTheGlobalFlag() {
+        assertDoesNotThrow(() -> guardWithAllowlist("10.0.0.1").requirePermitted("10.0.0.1"));
+        assertDoesNotThrow(() -> guardWithAllowlist("FILES.local").requirePermitted("files.LOCAL"));
+    }
+
+    @Test
+    void allowlistDoesNotOpenOtherPrivateHosts() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> guardWithAllowlist("10.0.0.1").requirePermitted("10.0.0.2"));
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/network/NetworkIdentitiesTest.java

@@ -0,0 +1,49 @@
+package stirling.software.proprietary.policy.network;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+class NetworkIdentitiesTest {
+
+    private static NetworkConfig config(String protocol, String share) {
+        Map<String, Object> options = new LinkedHashMap<>();
+        options.put("protocol", protocol);
+        options.put("host", "files.example.com");
+        options.put("port", "2222");
+        options.put("username", "u");
+        options.put("password", "p");
+        if (share != null) {
+            options.put("share", share);
+        }
+        return NetworkConfig.from(options);
+    }
+
+    @Test
+    void identityIsProtocolHostPortAndPath() {
+        assertEquals(
+                "sftp://files.example.com:2222/in/doc.pdf",
+                NetworkIdentities.identity(config("sftp", null), "in/doc.pdf"));
+    }
+
+    @Test
+    void smbIdentityIncludesTheShare() {
+        Map<String, Object> options = new LinkedHashMap<>();
+        options.put("protocol", "smb");
+        options.put("host", "files.example.com");
+        options.put("username", "u");
+        options.put("password", "p");
+        options.put("share", "documents");
+        assertEquals(
+                "smb://files.example.com:445/documents/reports/q1.pdf",
+                NetworkIdentities.identity(NetworkConfig.from(options), "reports/q1.pdf"));
+    }
+
+    @Test
+    void gateIsSizeAndModifiedTime() {
+        assertEquals("42:1700000000000", NetworkIdentities.gate(42, 1700000000000L));
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/network/NetworkInputSourceTest.java

@@ -0,0 +1,280 @@
+package stirling.software.proprietary.policy.network;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.access.service.OwnershipService;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.policy.input.ResolveContext;
+import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.security.service.UserService;
+
+/**
+ * Tests for {@link NetworkInputSource}: consume mode tracks remote files through the ledger, the
+ * version-guarded delete survives a mid-run replacement, a shared file is removed only once every
+ * policy is done, snapshot stays stateless, and the source type must match the connection protocol.
+ * The remote server is faked in memory, so no real SFTP/FTP/SMB endpoint is needed.
+ */
+class NetworkInputSourceTest {
+
+    private static final String POLICY = "p1";
+
+    private FakeServer server;
+    private NetworkInputSource source;
+    private InProcessProcessedLedger ledger;
+    private RecordingContext ctx;
+
+    @BeforeEach
+    void setUp() {
+        server = new FakeServer();
+        // allowPrivate=true so the host guard never resolves DNS; connect() is overridden anyway.
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateNetworkSources(true);
+        RemoteFileClientFactory factory =
+                new RemoteFileClientFactory(new NetworkHostGuard(properties)) {
+                    @Override
+                    public RemoteFileClient connect(NetworkConfig config) {
+                        return new FakeClient(server);
+                    }
+                };
+        NetworkConnectionResolver resolver =
+                new NetworkConnectionResolver(
+                        mock(IntegrationConfigRepository.class),
+                        mock(OwnershipService.class),
+                        mock(UserService.class));
+        source = new NetworkInputSource(resolver, factory);
+        ledger = new InProcessProcessedLedger();
+        ctx = new RecordingContext();
+    }
+
+    private static InputSpec sftp(Map<String, Object> extra) {
+        Map<String, Object> options = new LinkedHashMap<>();
+        options.put("protocol", "sftp");
+        options.put("host", "files.example.com");
+        options.put("username", "u");
+        options.put("password", "p");
+        options.put("directory", "in");
+        options.putAll(extra);
+        return new InputSpec("sftp", options);
+    }
+
+    @Test
+    void supportsTheThreeNetworkSourceTypes() {
+        assertTrue(source.supports(new InputSpec("sftp", Map.of())));
+        assertTrue(source.supports(new InputSpec("ftp", Map.of())));
+        assertTrue(source.supports(new InputSpec("network", Map.of())));
+        assertFalse(source.supports(new InputSpec("folder", Map.of())));
+    }
+
+    @Test
+    void consumeRemovesTheFileOnceProcessed() throws Exception {
+        server.put("in/doc.pdf", "data", 1000);
+
+        List<ResolvedInput> work = source.resolve(sftp(Map.of()), ctx);
+
+        assertEquals(1, work.size());
+        assertEquals(1, work.get(0).inputs().primary().size());
+        // In flight: still on the server, but a second sweep does not pick it up again.
+        assertTrue(server.has("in/doc.pdf"));
+        assertTrue(source.resolve(sftp(Map.of()), ctx).isEmpty());
+
+        work.get(0).onComplete().accept(true);
+        assertFalse(server.has("in/doc.pdf"));
+        assertTrue(source.resolve(sftp(Map.of()), ctx).isEmpty());
+    }
+
+    @Test
+    void aFileReplacedMidRunSurvivesTheDeleteAndRunsAgain() throws Exception {
+        server.put("in/doc.pdf", "data", 1000);
+
+        List<ResolvedInput> work = source.resolve(sftp(Map.of()), ctx);
+        // A new version lands while the run is executing (different size/mtime = new gate).
+        server.put("in/doc.pdf", "new data, different size", 2000);
+        work.get(0).onComplete().accept(true);
+
+        // The delete is version-guarded: the replacement is not the file that ran, so it stays and
+        // is claimed as fresh work next sweep.
+        assertTrue(server.has("in/doc.pdf"));
+        assertEquals(1, source.resolve(sftp(Map.of()), ctx).size());
+    }
+
+    @Test
+    void aSharedFileIsRemovedOnlyOnceEveryPolicyHasProcessedIt() throws Exception {
+        server.put("in/doc.pdf", "data", 1000);
+        RecordingContext other = new RecordingContext("p2");
+
+        List<ResolvedInput> mine = source.resolve(sftp(Map.of()), ctx);
+        List<ResolvedInput> theirs = source.resolve(sftp(Map.of()), other);
+        assertEquals(1, mine.size());
+        assertEquals(1, theirs.size());
+
+        mine.get(0).onComplete().accept(true);
+        assertTrue(server.has("in/doc.pdf")); // the other claim is still in flight
+
+        theirs.get(0).onComplete().accept(true);
+        assertFalse(server.has("in/doc.pdf"));
+    }
+
+    @Test
+    void aFailedFileStaysInPlaceAndIsNotRetriedUntilItChanges() throws Exception {
+        server.put("in/doc.pdf", "data", 1000);
+
+        source.resolve(sftp(Map.of()), ctx).get(0).onComplete().accept(false);
+
+        assertTrue(server.has("in/doc.pdf"));
+        assertTrue(source.resolve(sftp(Map.of()), ctx).isEmpty());
+
+        server.put("in/doc.pdf", "data", 2000); // touched: new mtime is a new version
+        assertEquals(1, source.resolve(sftp(Map.of()), ctx).size());
+    }
+
+    @Test
+    void snapshotReadsStatelesslyEverySweep() throws Exception {
+        server.put("in/doc.pdf", "data", 1000);
+        InputSpec spec = sftp(Map.of("mode", "snapshot"));
+
+        List<ResolvedInput> first = source.resolve(spec, ctx);
+        first.get(0).onComplete().accept(true);
+        List<ResolvedInput> second = source.resolve(spec, ctx);
+
+        assertEquals(1, first.size());
+        assertEquals(1, second.size()); // no ledger involvement
+        assertTrue(ctx.present.isEmpty());
+        assertTrue(server.has("in/doc.pdf"));
+    }
+
+    @Test
+    void streamsTheFileContent() throws Exception {
+        server.put("in/doc.pdf", "hello", 1000);
+
+        List<ResolvedInput> work = source.resolve(sftp(Map.of()), ctx);
+        try (InputStream in = work.get(0).inputs().primary().get(0).getInputStream()) {
+            assertEquals("hello", new String(in.readAllBytes(), StandardCharsets.UTF_8));
+        }
+    }
+
+    @Test
+    void validateRejectsASourceTypeThatDoesNotMatchTheConnectionProtocol() {
+        // A "network" (SMB) source pointed at an SFTP connection is a misconfiguration.
+        Map<String, Object> options = new LinkedHashMap<>();
+        options.put("protocol", "sftp");
+        options.put("host", "files.example.com");
+        options.put("username", "u");
+        options.put("password", "p");
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> source.validate(new InputSpec("network", options)));
+    }
+
+    /** In-memory remote server shared between a test's fake clients. */
+    private static final class FakeServer {
+        private final Map<String, FakeFile> files = new LinkedHashMap<>();
+
+        void put(String path, String content, long mtime) {
+            files.put(path, new FakeFile(content.getBytes(StandardCharsets.UTF_8), mtime));
+        }
+
+        boolean has(String path) {
+            return files.containsKey(path);
+        }
+    }
+
+    private record FakeFile(byte[] content, long mtime) {}
+
+    /** A RemoteFileClient over the in-memory server; every op reads/writes the shared map. */
+    private static final class FakeClient implements RemoteFileClient {
+        private final FakeServer server;
+
+        private FakeClient(FakeServer server) {
+            this.server = server;
+        }
+
+        @Override
+        public List<RemoteFile> list(String directory, boolean recursive) {
+            List<RemoteFile> out = new ArrayList<>();
+            server.files.forEach(
+                    (path, file) -> {
+                        String name = path.substring(path.lastIndexOf('/') + 1);
+                        out.add(new RemoteFile(path, name, file.content().length, file.mtime()));
+                    });
+            return out;
+        }
+
+        @Override
+        public RemoteFile stat(String path) {
+            FakeFile file = server.files.get(path);
+            if (file == null) {
+                return null;
+            }
+            String name = path.substring(path.lastIndexOf('/') + 1);
+            return new RemoteFile(path, name, file.content().length, file.mtime());
+        }
+
+        @Override
+        public InputStream open(String path) {
+            return new ByteArrayInputStream(server.files.get(path).content());
+        }
+
+        @Override
+        public void delete(String path) {
+            server.files.remove(path);
+        }
+
+        @Override
+        public void close() {}
+    }
+
+    /** Policy-scoped context backed by the in-process ledger, recording presence reports. */
+    private final class RecordingContext implements ResolveContext {
+        private final String policyId;
+        private final List<String> present = new ArrayList<>();
+
+        private RecordingContext() {
+            this(POLICY);
+        }
+
+        private RecordingContext(String policyId) {
+            this.policyId = policyId;
+        }
+
+        @Override
+        public boolean claim(String identity, String gate, Supplier<String> contentHash) {
+            return ledger.claim(policyId, identity, gate, contentHash);
+        }
+
+        @Override
+        public void settle(
+                String identity, String finalGate, String finalContentHash, boolean success) {
+            ledger.settle(policyId, identity, finalGate, finalContentHash, success);
+        }
+
+        @Override
+        public boolean allSettledDone(String identity) {
+            return ledger.allSettledDone(identity);
+        }
+
+        @Override
+        public void reportPresent(Collection<String> identities) {
+            present.addAll(identities);
+        }
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/network/NetworkIntegrationValidatorTest.java

@@ -0,0 +1,55 @@
+package stirling.software.proprietary.policy.network;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.integration.model.IntegrationType;
+
+class NetworkIntegrationValidatorTest {
+
+    private static NetworkIntegrationValidator validator(boolean allowPrivate) {
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateNetworkSources(allowPrivate);
+        return new NetworkIntegrationValidator(new NetworkHostGuard(properties));
+    }
+
+    private static Map<String, Object> config(String host) {
+        Map<String, Object> options = new LinkedHashMap<>();
+        options.put("protocol", "sftp");
+        options.put("host", host);
+        options.put("username", "u");
+        options.put("password", "p");
+        return options;
+    }
+
+    @Test
+    void reportsTheNetworkType() {
+        assertEquals(IntegrationType.NETWORK, validator(true).type());
+    }
+
+    @Test
+    void acceptsAValidConfig() {
+        assertDoesNotThrow(() -> validator(true).validate(config("10.0.0.5")));
+    }
+
+    @Test
+    void rejectsAMalformedConfig() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> validator(true).validate(Map.of("host", "h")));
+    }
+
+    @Test
+    void rejectsAPrivateHostUnlessOptedIn() {
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> validator(false).validate(config("10.0.0.5")));
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/network/SftpNetworkSourceIntegrationTest.java

@@ -0,0 +1,241 @@
+package stirling.software.proprietary.policy.network;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import stirling.software.common.configuration.InstallationPathConfig;
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.access.service.OwnershipService;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.policy.input.ResolveContext;
+import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.security.service.UserService;
+
+/**
+ * End-to-end {@link NetworkInputSource} test over SFTP against a real atmoz/sftp server, through
+ * the production {@link SftpFileClient}: listing, claiming, streaming the actual bytes, consume
+ * delete, and save-time validation. The server chroots the user to its home; {@code upload} is the
+ * writable drop folder.
+ */
+@Testcontainers(disabledWithoutDocker = true)
+@EnabledIfEnvironmentVariable(
+        named = "RUN_NETWORK_INTEGRATION_TESTS",
+        matches = "true",
+        disabledReason =
+                "Spins up SFTP/FTP/SMB containers; opt-in to keep several heavy containers off the"
+                        + " standard CI runner. Run with RUN_NETWORK_INTEGRATION_TESTS=true.")
+class SftpNetworkSourceIntegrationTest {
+
+    private static final String POLICY = "p1";
+    private static final String USER = "stirling";
+    private static final String PASS = "secret";
+    private static final String DIR = "upload";
+    private static final int SFTP_PORT = 22;
+
+    @org.testcontainers.junit.jupiter.Container
+    static GenericContainer<?> sftp =
+            new GenericContainer<>("atmoz/sftp:alpine")
+                    .withExposedPorts(SFTP_PORT)
+                    .withCommand(USER + ":" + PASS + ":::" + DIR)
+                    .waitingFor(Wait.forListeningPort());
+
+    private NetworkInputSource source;
+    private InProcessProcessedLedger ledger;
+    private RecordingContext ctx;
+
+    @BeforeEach
+    void setUp() throws IOException {
+        // Each run's container has a fresh host key, so drop pins from earlier runs: a reused
+        // mapped port would otherwise trip the trust-on-first-use "key changed" refusal.
+        Files.deleteIfExists(
+                Path.of(InstallationPathConfig.getConfigPath(), SftpFileClient.KNOWN_HOSTS_FILE));
+        // The container resolves to loopback, so the private-host opt-in must be on.
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateNetworkSources(true);
+        RemoteFileClientFactory factory =
+                new RemoteFileClientFactory(new NetworkHostGuard(properties));
+        NetworkConnectionResolver resolver =
+                new NetworkConnectionResolver(
+                        mock(IntegrationConfigRepository.class),
+                        mock(OwnershipService.class),
+                        mock(UserService.class));
+        source = new NetworkInputSource(resolver, factory);
+        ledger = new InProcessProcessedLedger();
+        ctx = new RecordingContext();
+    }
+
+    @Test
+    void consumeListsStreamsAndDeletes() throws Exception {
+        put("doc.pdf", "hello sftp");
+
+        List<ResolvedInput> work = source.resolve(spec(Map.of()), ctx);
+
+        assertThat(work).hasSize(1);
+        String identity =
+                "sftp://"
+                        + sftp.getHost()
+                        + ":"
+                        + sftp.getMappedPort(SFTP_PORT)
+                        + "/upload/doc.pdf";
+        assertThat(ctx.present).containsExactly(identity);
+        assertThat(read(work.get(0))).isEqualTo("hello sftp");
+        // In flight: a second sweep does not re-claim it.
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+
+        work.get(0).onComplete().accept(true);
+        assertThat(exists("doc.pdf")).isFalse();
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+    }
+
+    @Test
+    void aFailedFileStaysOnTheServer() throws Exception {
+        put("doc.pdf", "data");
+
+        source.resolve(spec(Map.of()), ctx).get(0).onComplete().accept(false);
+
+        assertThat(exists("doc.pdf")).isTrue();
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+    }
+
+    @Test
+    void validateConnectsAndRejectsBadCredentials() {
+        source.validate(spec(Map.of()));
+
+        Map<String, Object> wrong = new HashMap<>(baseOptions());
+        wrong.put("password", "not-the-password");
+        assertThatThrownBy(() -> source.validate(new InputSpec("sftp", wrong)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("cannot access");
+    }
+
+    @Test
+    void acceptsTheServersRealHostKeyFingerprint() throws Exception {
+        String fingerprint =
+                exec("ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub | awk '{print $2}'").trim();
+        Map<String, Object> pinned = new HashMap<>(baseOptions());
+        pinned.put("hostKeyFingerprint", fingerprint);
+        source.validate(new InputSpec("sftp", pinned));
+    }
+
+    @Test
+    void rejectsAWrongHostKeyFingerprint() {
+        Map<String, Object> wrong = new HashMap<>(baseOptions());
+        wrong.put("hostKeyFingerprint", "SHA256:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA");
+        // Refused during the handshake, before any credentials are sent.
+        assertThatThrownBy(() -> source.validate(new InputSpec("sftp", wrong)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("cannot access");
+    }
+
+    @Test
+    void refusesAChangedHostKey() throws Exception {
+        source.validate(spec(Map.of())); // pins the current key (trust-on-first-use)
+
+        Path knownHosts =
+                Path.of(InstallationPathConfig.getConfigPath(), SftpFileClient.KNOWN_HOSTS_FILE);
+        String pinned = Files.readString(knownHosts);
+        // Swap the pinned key for a valid-but-different one so the server's real key reads as
+        // changed rather than unparseable.
+        Files.writeString(knownHosts, pinned.replaceFirst("(?m) [^ ]+$", " " + bogusEd25519Key()));
+
+        assertThatThrownBy(() -> source.validate(spec(Map.of())))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("cannot access");
+    }
+
+    /** A structurally valid ssh-ed25519 public key blob (all-zero key material), base64. */
+    private static String bogusEd25519Key() {
+        byte[] type = "ssh-ed25519".getBytes(StandardCharsets.US_ASCII);
+        ByteBuffer blob = ByteBuffer.allocate(4 + type.length + 4 + 32);
+        blob.putInt(type.length).put(type).putInt(32);
+        return java.util.Base64.getEncoder().encodeToString(blob.array());
+    }
+
+    private Map<String, Object> baseOptions() {
+        Map<String, Object> options = new HashMap<>();
+        options.put("protocol", "sftp");
+        options.put("host", sftp.getHost());
+        options.put("port", String.valueOf(sftp.getMappedPort(SFTP_PORT)));
+        options.put("username", USER);
+        options.put("password", PASS);
+        options.put("directory", DIR);
+        return options;
+    }
+
+    private InputSpec spec(Map<String, Object> extra) {
+        Map<String, Object> options = new HashMap<>(baseOptions());
+        options.putAll(extra);
+        return new InputSpec("sftp", options);
+    }
+
+    private void put(String name, String content) throws Exception {
+        String path = "/home/" + USER + "/" + DIR + "/" + name;
+        exec("printf '%s' '" + content + "' > " + path + " && chmod 644 " + path);
+    }
+
+    private boolean exists(String name) throws Exception {
+        return exec("test -f /home/" + USER + "/" + DIR + "/" + name + " && echo yes || echo no")
+                .contains("yes");
+    }
+
+    private String exec(String script) throws Exception {
+        Container.ExecResult result = sftp.execInContainer("sh", "-c", script);
+        return result.getStdout() + result.getStderr();
+    }
+
+    private static String read(ResolvedInput unit) throws IOException {
+        try (InputStream stream = unit.inputs().primary().get(0).getInputStream()) {
+            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
+        }
+    }
+
+    private class RecordingContext implements ResolveContext {
+
+        private final List<String> present = new ArrayList<>();
+
+        @Override
+        public boolean claim(String identity, String gate, Supplier<String> contentHash) {
+            return ledger.claim(POLICY, identity, gate, contentHash);
+        }
+
+        @Override
+        public void settle(
+                String identity, String finalGate, String finalContentHash, boolean success) {
+            ledger.settle(POLICY, identity, finalGate, finalContentHash, success);
+        }
+
+        @Override
+        public boolean allSettledDone(String identity) {
+            return ledger.allSettledDone(identity);
+        }
+
+        @Override
+        public void reportPresent(Collection<String> identities) {
+            present.addAll(identities);
+        }
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/network/SmbNetworkSourceIntegrationTest.java

@@ -0,0 +1,194 @@
+package stirling.software.proprietary.policy.network;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.access.service.OwnershipService;
+import stirling.software.proprietary.integration.repository.IntegrationConfigRepository;
+import stirling.software.proprietary.policy.input.ResolveContext;
+import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.security.service.UserService;
+
+/**
+ * End-to-end {@link NetworkInputSource} test over SMB against a real Samba server, through the
+ * production {@link SmbFileClient}: listing a share, claiming, streaming the actual bytes, consume
+ * delete, and save-time validation. The share is served world-writable so the SMB user can remove a
+ * consumed file.
+ */
+@Testcontainers(disabledWithoutDocker = true)
+@EnabledIfEnvironmentVariable(
+        named = "RUN_NETWORK_INTEGRATION_TESTS",
+        matches = "true",
+        disabledReason =
+                "Spins up SFTP/FTP/SMB containers; opt-in to keep several heavy containers off the"
+                        + " standard CI runner. Run with RUN_NETWORK_INTEGRATION_TESTS=true.")
+class SmbNetworkSourceIntegrationTest {
+
+    private static final String POLICY = "p1";
+    private static final String USER = "stirling";
+    private static final String PASS = "secret";
+    private static final String SHARE = "documents";
+    private static final int SMB_PORT = 445;
+
+    @org.testcontainers.junit.jupiter.Container
+    static GenericContainer<?> samba =
+            new GenericContainer<>("dperson/samba")
+                    .withExposedPorts(SMB_PORT)
+                    .withCommand(
+                            "-p",
+                            "-u",
+                            USER + ";" + PASS,
+                            "-s",
+                            SHARE + ";/share;yes;no;no;" + USER)
+                    .waitingFor(Wait.forListeningPort());
+
+    private NetworkInputSource source;
+    private InProcessProcessedLedger ledger;
+    private RecordingContext ctx;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        // The share dir must be writable by the SMB user so a consumed file can be deleted.
+        samba.execInContainer("sh", "-c", "mkdir -p /share && chmod -R 0777 /share");
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateNetworkSources(true);
+        RemoteFileClientFactory factory =
+                new RemoteFileClientFactory(new NetworkHostGuard(properties));
+        NetworkConnectionResolver resolver =
+                new NetworkConnectionResolver(
+                        mock(IntegrationConfigRepository.class),
+                        mock(OwnershipService.class),
+                        mock(UserService.class));
+        source = new NetworkInputSource(resolver, factory);
+        ledger = new InProcessProcessedLedger();
+        ctx = new RecordingContext();
+    }
+
+    @Test
+    void consumeListsStreamsAndDeletes() throws Exception {
+        put("doc.pdf", "hello smb");
+
+        List<ResolvedInput> work = source.resolve(spec(Map.of()), ctx);
+
+        assertThat(work).hasSize(1);
+        String identity =
+                "smb://"
+                        + samba.getHost()
+                        + ":"
+                        + samba.getMappedPort(SMB_PORT)
+                        + "/documents/doc.pdf";
+        assertThat(ctx.present).containsExactly(identity);
+        assertThat(read(work.get(0))).isEqualTo("hello smb");
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+
+        work.get(0).onComplete().accept(true);
+        assertThat(exists("doc.pdf")).isFalse();
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+    }
+
+    @Test
+    void aFailedFileStaysOnTheShare() throws Exception {
+        put("doc.pdf", "data");
+
+        source.resolve(spec(Map.of()), ctx).get(0).onComplete().accept(false);
+
+        assertThat(exists("doc.pdf")).isTrue();
+        assertThat(source.resolve(spec(Map.of()), ctx)).isEmpty();
+    }
+
+    @Test
+    void validateConnectsAndRejectsBadCredentials() {
+        source.validate(spec(Map.of()));
+
+        Map<String, Object> wrong = new HashMap<>(baseOptions());
+        wrong.put("password", "not-the-password");
+        assertThatThrownBy(() -> source.validate(new InputSpec("network", wrong)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("cannot access");
+    }
+
+    private Map<String, Object> baseOptions() {
+        Map<String, Object> options = new HashMap<>();
+        options.put("protocol", "smb");
+        options.put("host", samba.getHost());
+        options.put("port", String.valueOf(samba.getMappedPort(SMB_PORT)));
+        options.put("username", USER);
+        options.put("password", PASS);
+        options.put("share", SHARE);
+        return options;
+    }
+
+    private InputSpec spec(Map<String, Object> extra) {
+        Map<String, Object> options = new HashMap<>(baseOptions());
+        options.putAll(extra);
+        return new InputSpec("network", options);
+    }
+
+    private void put(String name, String content) throws Exception {
+        String path = "/share/" + name;
+        exec("printf '%s' '" + content + "' > " + path + " && chmod 0666 " + path);
+    }
+
+    private boolean exists(String name) throws Exception {
+        return exec("test -f /share/" + name + " && echo yes || echo no").contains("yes");
+    }
+
+    private String exec(String script) throws Exception {
+        Container.ExecResult result = samba.execInContainer("sh", "-c", script);
+        return result.getStdout() + result.getStderr();
+    }
+
+    private static String read(ResolvedInput unit) throws IOException {
+        try (InputStream stream = unit.inputs().primary().get(0).getInputStream()) {
+            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
+        }
+    }
+
+    private class RecordingContext implements ResolveContext {
+
+        private final List<String> present = new ArrayList<>();
+
+        @Override
+        public boolean claim(String identity, String gate, Supplier<String> contentHash) {
+            return ledger.claim(POLICY, identity, gate, contentHash);
+        }
+
+        @Override
+        public void settle(
+                String identity, String finalGate, String finalContentHash, boolean success) {
+            ledger.settle(POLICY, identity, finalGate, finalContentHash, success);
+        }
+
+        @Override
+        public boolean allSettledDone(String identity) {
+            return ledger.allSettledDone(identity);
+        }
+
+        @Override
+        public void reportPresent(Collection<String> identities) {
+            present.addAll(identities);
+        }
+    }
+}

build.gradle

@@ -44,6 +44,9 @@ ext {
     jpdfiumVersion = "1.0.2"
     jwtVersion = "0.13.0"
     awsSdkVersion = "2.44.12"
+    jschVersion = "0.2.23"
+    commonsNetVersion = "3.11.1"
+    smbjVersion = "0.14.0"
     tinkVersion = "1.23.0"
     testcontainersMinioVersion = "1.21.4"
     // junit-platform-launcher version managed by Spring Boot BOM

frontend/editor/public/locales/en-US/translation.toml

@@ -6701,9 +6701,15 @@ label = "Server URL"
 [portal.connections.commonFields.email]
 label = "Email"
 
+[portal.connections.commonFields.host]
+label = "Host"
+
 [portal.connections.commonFields.password]
 label = "Password"
 
+[portal.connections.commonFields.port]
+label = "Port"
+
 [portal.connections.commonFields.username]
 label = "Username"
 
@@ -6836,6 +6842,27 @@ baseUrlPlaceholder = "https://your-cluster:9200"
 description = "Index an audit event in Elasticsearch each time a policy handles a document."
 label = "Elasticsearch"
 
+[portal.connections.types.ftp]
+description = "Poll an FTP or FTPS server for new documents."
+hostPlaceholder = "ftp.example.com"
+label = "FTP / FTPS"
+
+[portal.connections.types.ftp.fields.passive]
+label = "Connection mode"
+
+[portal.connections.types.ftp.fields.passive.options]
+active = "Active"
+passive = "Passive (recommended)"
+
+[portal.connections.types.ftp.fields.security]
+helperText = "Plain FTP sends credentials and files unencrypted; choose FTPS if the server supports it."
+label = "Encryption"
+
+[portal.connections.types.ftp.fields.security.options]
+explicit = "FTPS (explicit / AUTH TLS)"
+implicit = "FTPS (implicit)"
+none = "None (plain FTP)"
+
 [portal.connections.types.googlechat]
 baseUrlPlaceholder = "https://chat.googleapis.com/v1/spaces/..."
 description = "Post a message to a Google Chat space when a policy runs."
@@ -6903,10 +6930,43 @@ label = "Secret access key"
 description = "Email a document or a notification when a policy runs."
 label = "SendGrid"
 
+[portal.connections.types.sftp]
+description = "Poll an SFTP (SSH) server for new documents."
+hostPlaceholder = "sftp.example.com"
+label = "SFTP"
+
+[portal.connections.types.sftp.fields.hostKeyFingerprint]
+helperText = "Optional. The server's SHA-256 host key fingerprint (from `ssh-keyscan host | ssh-keygen -lf -`). When set, only a server presenting this key is accepted; when blank, the key seen on first connect is pinned and later changes are refused."
+label = "Host key fingerprint"
+placeholder = "SHA256:..."
+
+[portal.connections.types.sftp.fields.passphrase]
+label = "Key passphrase"
+
+[portal.connections.types.sftp.fields.password]
+helperText = "Leave blank if you authenticate with a private key instead."
+
+[portal.connections.types.sftp.fields.privateKey]
+helperText = "Optional. Paste an OpenSSH private key to authenticate with a key instead of a password."
+label = "Private key"
+
 [portal.connections.types.slack]
 description = "Post a message to Slack when a policy processes a document."
 label = "Slack"
 
+[portal.connections.types.smb]
+description = "Watch an SMB / CIFS network share (Windows or Samba)."
+hostPlaceholder = "fileserver.example.com"
+label = "SMB / Network share"
+
+[portal.connections.types.smb.fields.domain]
+helperText = "Optional. The Windows domain or workgroup, if your server uses one."
+label = "Domain"
+
+[portal.connections.types.smb.fields.share]
+label = "Share"
+placeholder = "documents"
+
 [portal.connections.types.splunk]
 baseUrlPlaceholder = "https://your-splunk:8088"
 description = "Send an audit event to Splunk each time a policy handles a document."
@@ -8711,6 +8771,30 @@ inUse = "In use"
 total = "Connections"
 unused = "Unused"
 
+[portal.sources.networkFields.connection]
+helperText = "The stored connection with the server address and credentials. Reused by every source that references it."
+label = "Connection"
+
+[portal.sources.networkFields.directory]
+helperText = "Folder on the server to poll, relative to the login home or share root. Leave blank for the root."
+label = "Folder"
+placeholder = "incoming/"
+
+[portal.sources.networkFields.mode]
+helperText = "Consume removes each file from the server once every policy has processed it."
+label = "Read mode"
+
+[portal.sources.networkFields.mode.options]
+consume = "Consume: process each file once"
+snapshot = "Snapshot: re-read the folder every run"
+
+[portal.sources.networkFields.recursive]
+label = "Folder depth"
+
+[portal.sources.networkFields.recursive.options]
+all = "Include subfolders"
+top = "Top level only"
+
 [portal.sources.status]
 active = "Active"
 disabled = "Disabled"
@@ -8770,6 +8854,10 @@ label = "Folder depth"
 all = "Include subfolders"
 top = "Top level only"
 
+[portal.sources.types.ftp]
+description = "Poll an FTP or FTPS server folder for new documents."
+label = "FTP"
+
 [portal.sources.types.googledrive]
 description = "Pull documents from shared Google Drive folders."
 label = "Google Drive"

frontend/editor/src/portal/api/integrations.ts

@@ -6,7 +6,13 @@
  */
 import { apiClient } from "@portal/api/http";
 
-export type IntegrationType = "S3" | "MCP" | "API" | "PURVIEW" | "CONSIGNO";
+export type IntegrationType =
+  | "S3"
+  | "NETWORK"
+  | "MCP"
+  | "API"
+  | "PURVIEW"
+  | "CONSIGNO";
 export type OwnerScope = "USER" | "TEAM" | "SERVER";
 
 /** Mirrors the backend IntegrationConfigResponse; `config` values are masked. */

frontend/editor/src/portal/components/BrandMarks.tsx

@@ -383,6 +383,14 @@ const NEUTRAL: Record<string, ReactNode> = {
       <path d="m8.5 6.5 3.5-3 3.5 3" />
     </>
   ),
+  ftp: (
+    <>
+      <rect x="4" y="13" width="16" height="7" rx="1.5" />
+      <path d="M7.5 16.5h.01" />
+      <path d="M11 16.5h.01" />
+      <path d="M9 4.5h6l-2-2M15 8.5H9l2 2" />
+    </>
+  ),
   email: (
     <>
       <path d="M21 12.5V18a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2v-5.5" />

frontend/editor/src/portal/components/sources/ConnectionForm.tsx

@@ -26,8 +26,24 @@ export function ConnectionForm({
   onChange,
 }: ConnectionFormProps) {
   const { t } = useTranslation();
-  const set = (key: string, value: string) =>
-    onChange({ ...values, [key]: value });
+  const set = (key: string, value: string) => {
+    const next = { ...values, [key]: value };
+    const sync = type.fields.find((f) => f.key === key)?.syncsDefault;
+    if (sync) {
+      const target = next[sync.targetKey] ?? "";
+      const targetDefault =
+        type.fields.find((f) => f.key === sync.targetKey)?.defaultValue ?? "";
+      const untouched =
+        target === "" ||
+        target === targetDefault ||
+        Object.values(sync.map).includes(target);
+      const mapped = sync.map[value];
+      if (untouched && mapped !== undefined) {
+        next[sync.targetKey] = mapped;
+      }
+    }
+    onChange(next);
+  };
 
   return (
     <div className="portal-sources__connection-form">

frontend/editor/src/portal/components/sources/SourceModal.tsx

@@ -32,12 +32,14 @@ import {
 } from "@portal/components/sources/sourceTypes";
 import { BrandMark } from "@portal/components/BrandMarks";
 import { S3ConnectionPicker } from "@portal/components/sources/S3ConnectionPicker";
+import { ConnectionPicker } from "@portal/components/sources/ConnectionPicker";
 import { ConnectionForm } from "@portal/components/sources/ConnectionForm";
 import {
   CREATABLE_CONNECTION_TYPES,
   buildConnectionConfig,
   connectionFormValid,
   emptyConnectionValues,
+  type CreatableConnectionType,
 } from "@portal/components/sources/connectionTypes";
 import { createIntegration } from "@portal/api/integrations";
 import "@portal/components/sources/SourceModal.css";
@@ -76,6 +78,14 @@ const S3_CONNECTION_TYPE = CREATABLE_CONNECTION_TYPES.find(
   (entry) => entry.id === "s3",
 )!;
 
+/** A connection-catalogue entry by id, for the inline create form a source field opens. */
+function connectionTypeById(id: string): CreatableConnectionType {
+  return (
+    CREATABLE_CONNECTION_TYPES.find((entry) => entry.id === id) ??
+    S3_CONNECTION_TYPE
+  );
+}
+
 interface SourceModalProps {
   open: boolean;
   /** When set, edit this source; otherwise create a new one. */
@@ -129,7 +139,10 @@ export function SourceModal({
     webhookId: string;
     secret: string;
   } | null>(null);
-  // In-place connection create (swaps the stage; never stacks a second modal).
+  // In-place connection create (swaps the stage; never stacks a second modal). The type is set
+  // per field, so a source that wants an SFTP connection opens the SFTP form, not always S3.
+  const [connType, setConnType] =
+    useState<CreatableConnectionType>(S3_CONNECTION_TYPE);
   const [connValues, setConnValues] = useState<Record<string, string>>(() =>
     emptyConnectionValues(S3_CONNECTION_TYPE),
   );
@@ -192,24 +205,27 @@ export function SourceModal({
     onClose();
   }
 
-  function openConnectionStage(fieldKey: string) {
-    setConnValues(emptyConnectionValues(S3_CONNECTION_TYPE));
+  function openConnectionStage(
+    fieldKey: string,
+    type: CreatableConnectionType,
+  ) {
+    setConnType(type);
+    setConnValues(emptyConnectionValues(type));
     setConnField(fieldKey);
     setError(null);
     setStage("connection");
   }
 
   async function saveConnection() {
-    if (connSaving || !connectionFormValid(S3_CONNECTION_TYPE, connValues))
-      return;
+    if (connSaving || !connectionFormValid(connType, connValues)) return;
     setConnSaving(true);
     setError(null);
     try {
       const created = await createIntegration({
-        integrationType: S3_CONNECTION_TYPE.integrationType,
+        integrationType: connType.integrationType,
         name: connValues.name.trim(),
         scope: "TEAM",
-        config: buildConnectionConfig(S3_CONNECTION_TYPE, connValues),
+        config: buildConnectionConfig(connType, connValues),
       });
       // Back to the source form with the fresh connection selected; the picker
       // remounts and refetches, so the new name is in its list.
@@ -277,7 +293,7 @@ export function SourceModal({
       ? t("portal.sources.builder.createTitle")
       : stage === "connection"
         ? t("portal.connections.createTitleFor", {
-            name: t(S3_CONNECTION_TYPE.labelKey),
+            name: t(connType.labelKey),
           })
         : stage === "reveal"
           ? t("portal.sources.types.webhook.reveal.title")
@@ -346,7 +362,7 @@ export function SourceModal({
             <Button
               size="sm"
               loading={connSaving}
-              disabled={!connectionFormValid(S3_CONNECTION_TYPE, connValues)}
+              disabled={!connectionFormValid(connType, connValues)}
               onClick={() => void saveConnection()}
             >
               {t("portal.connections.picker.save")}
@@ -512,7 +528,28 @@ export function SourceModal({
                       onChange={(connectionId) =>
                         setOption(field.key, connectionId)
                       }
-                      onCreateNew={() => openConnectionStage(field.key)}
+                      onCreateNew={() =>
+                        openConnectionStage(field.key, S3_CONNECTION_TYPE)
+                      }
+                    />
+                  ) : field.control === "connection" ? (
+                    <ConnectionPicker
+                      value={options[field.key] ?? ""}
+                      onChange={(connectionId) =>
+                        setOption(field.key, connectionId)
+                      }
+                      integrationType={
+                        connectionTypeById(field.connectionTypeId ?? "")
+                          .integrationType
+                      }
+                      createTypeId={field.connectionTypeId ?? ""}
+                      presetId={field.connectionTypeId}
+                      onCreateNew={() =>
+                        openConnectionStage(
+                          field.key,
+                          connectionTypeById(field.connectionTypeId ?? ""),
+                        )
+                      }
                     />
                   ) : field.control === "select" ? (
                     <Select
@@ -593,7 +630,7 @@ export function SourceModal({
       {stage === "connection" && (
         <div className="portal-source-modal__form">
           <ConnectionForm
-            type={S3_CONNECTION_TYPE}
+            type={connType}
             values={connValues}
             onChange={setConnValues}
           />

frontend/editor/src/portal/components/sources/connectionTypes.ts

@@ -29,6 +29,12 @@ export interface ConnectionFieldDef {
   defaultValue?: string;
   /** Shown only when another field has one of these values, e.g. auth fields per authType. */
   visibleWhen?: { key: string; oneOf: string[] };
+  /**
+   * When this field changes, move another field onto the default paired with the new value (FTP
+   * port per encryption mode) — but only while the target still holds a default, never a custom
+   * value the operator typed.
+   */
+  syncsDefault?: { targetKey: string; map: Record<string, string> };
 }
 
 export interface CreatableConnectionType {
@@ -122,6 +128,162 @@ const S3_FIELDS: ConnectionFieldDef[] = [
   },
 ];
 
+// Network file servers (SFTP/FTP/SMB). The protocol is baked into presetConfig; the operator
+// supplies host and credentials. host/port/username/password reuse the shared commonFields copy.
+const SFTP_FIELDS: ConnectionFieldDef[] = [
+  {
+    key: "host",
+    labelKey: `${COMMON}.host.label`,
+    control: "text",
+    required: true,
+    placeholderKey: `${PREFIX}.sftp.hostPlaceholder`,
+  },
+  {
+    key: "port",
+    labelKey: `${COMMON}.port.label`,
+    control: "text",
+    defaultValue: "22",
+  },
+  {
+    key: "username",
+    labelKey: `${COMMON}.username.label`,
+    control: "text",
+    required: true,
+  },
+  {
+    key: "password",
+    labelKey: `${COMMON}.password.label`,
+    control: "password",
+    helperTextKey: `${PREFIX}.sftp.fields.password.helperText`,
+  },
+  {
+    key: "privateKey",
+    labelKey: `${PREFIX}.sftp.fields.privateKey.label`,
+    control: "textarea",
+    helperTextKey: `${PREFIX}.sftp.fields.privateKey.helperText`,
+  },
+  {
+    key: "privateKeyPassphrase",
+    labelKey: `${PREFIX}.sftp.fields.passphrase.label`,
+    control: "password",
+  },
+  {
+    key: "hostKeyFingerprint",
+    labelKey: `${PREFIX}.sftp.fields.hostKeyFingerprint.label`,
+    control: "text",
+    placeholderKey: `${PREFIX}.sftp.fields.hostKeyFingerprint.placeholder`,
+    helperTextKey: `${PREFIX}.sftp.fields.hostKeyFingerprint.helperText`,
+  },
+];
+
+const FTP_FIELDS: ConnectionFieldDef[] = [
+  {
+    key: "host",
+    labelKey: `${COMMON}.host.label`,
+    control: "text",
+    required: true,
+    placeholderKey: `${PREFIX}.ftp.hostPlaceholder`,
+  },
+  {
+    key: "port",
+    labelKey: `${COMMON}.port.label`,
+    control: "text",
+    defaultValue: "21",
+  },
+  {
+    key: "username",
+    labelKey: `${COMMON}.username.label`,
+    control: "text",
+    required: true,
+  },
+  {
+    key: "password",
+    labelKey: `${COMMON}.password.label`,
+    control: "password",
+    required: true,
+  },
+  {
+    key: "security",
+    labelKey: `${PREFIX}.ftp.fields.security.label`,
+    control: "select",
+    defaultValue: "NONE",
+    helperTextKey: `${PREFIX}.ftp.fields.security.helperText`,
+    // Implicit FTPS listens on 990; follow the untouched port default across modes.
+    syncsDefault: {
+      targetKey: "port",
+      map: { NONE: "21", EXPLICIT: "21", IMPLICIT: "990" },
+    },
+    options: [
+      { value: "NONE", labelKey: `${PREFIX}.ftp.fields.security.options.none` },
+      {
+        value: "EXPLICIT",
+        labelKey: `${PREFIX}.ftp.fields.security.options.explicit`,
+      },
+      {
+        value: "IMPLICIT",
+        labelKey: `${PREFIX}.ftp.fields.security.options.implicit`,
+      },
+    ],
+  },
+  {
+    key: "passive",
+    labelKey: `${PREFIX}.ftp.fields.passive.label`,
+    control: "select",
+    defaultValue: "true",
+    options: [
+      {
+        value: "true",
+        labelKey: `${PREFIX}.ftp.fields.passive.options.passive`,
+      },
+      {
+        value: "false",
+        labelKey: `${PREFIX}.ftp.fields.passive.options.active`,
+      },
+    ],
+  },
+];
+
+const SMB_FIELDS: ConnectionFieldDef[] = [
+  {
+    key: "host",
+    labelKey: `${COMMON}.host.label`,
+    control: "text",
+    required: true,
+    placeholderKey: `${PREFIX}.smb.hostPlaceholder`,
+  },
+  {
+    key: "port",
+    labelKey: `${COMMON}.port.label`,
+    control: "text",
+    defaultValue: "445",
+  },
+  {
+    key: "share",
+    labelKey: `${PREFIX}.smb.fields.share.label`,
+    control: "text",
+    required: true,
+    placeholderKey: `${PREFIX}.smb.fields.share.placeholder`,
+  },
+  {
+    key: "username",
+    labelKey: `${COMMON}.username.label`,
+    control: "text",
+    required: true,
+  },
+  {
+    key: "password",
+    labelKey: `${COMMON}.password.label`,
+    control: "password",
+    required: true,
+  },
+  {
+    key: "domain",
+    labelKey: `${PREFIX}.smb.fields.domain.label`,
+    control: "text",
+    helperTextKey: `${PREFIX}.smb.fields.domain.helperText`,
+  },
+];
+
 const PURVIEW_FIELDS: ConnectionFieldDef[] = [
   {
     key: "tenantId",
@@ -587,6 +749,46 @@ export const CREATABLE_CONNECTION_TYPES: CreatableConnectionType[] = [
     searchTerms: ["aws", "bucket", "minio", "object storage"],
     fields: S3_FIELDS,
   },
+  {
+    id: "sftp",
+    integrationType: "NETWORK",
+    kind: "preset",
+    category: "storage",
+    labelKey: `${PREFIX}.sftp.label`,
+    descriptionKey: `${PREFIX}.sftp.description`,
+    searchTerms: ["sftp", "ssh", "scp", "drop folder", "file transfer"],
+    presetConfig: { protocol: "SFTP" },
+    fields: SFTP_FIELDS,
+  },
+  {
+    id: "ftp",
+    integrationType: "NETWORK",
+    kind: "preset",
+    category: "storage",
+    labelKey: `${PREFIX}.ftp.label`,
+    descriptionKey: `${PREFIX}.ftp.description`,
+    searchTerms: ["ftp", "ftps", "file transfer", "drop folder"],
+    presetConfig: { protocol: "FTP" },
+    fields: FTP_FIELDS,
+  },
+  {
+    id: "smb",
+    integrationType: "NETWORK",
+    kind: "preset",
+    category: "storage",
+    labelKey: `${PREFIX}.smb.label`,
+    descriptionKey: `${PREFIX}.smb.description`,
+    searchTerms: [
+      "smb",
+      "cifs",
+      "samba",
+      "network drive",
+      "windows share",
+      "unc",
+    ],
+    presetConfig: { protocol: "SMB" },
+    fields: SMB_FIELDS,
+  },
   {
     id: "purview",
     integrationType: "PURVIEW",

frontend/editor/src/portal/components/sources/sourceTypes.ts

@@ -37,6 +37,18 @@ const SOURCE_TYPE_META: Record<string, SourceTypeMeta> = {
     labelKey: "portal.sources.types.s3.label",
     accent: "brand",
   },
+  sftp: {
+    labelKey: "portal.sources.types.sftp.label",
+    accent: "default",
+  },
+  ftp: {
+    labelKey: "portal.sources.types.ftp.label",
+    accent: "default",
+  },
+  network: {
+    labelKey: "portal.sources.types.network.label",
+    accent: "default",
+  },
   webhook: {
     labelKey: "portal.sources.types.webhook.label",
     accent: "warning",
@@ -56,12 +68,17 @@ export function sourceTypeMeta(type: string): SourceTypeMeta {
 export interface SourceFieldDef {
   key: string;
   labelKey: string;
-  control: "text" | "password" | "select" | "s3Connection";
+  control: "text" | "password" | "select" | "s3Connection" | "connection";
   required?: boolean;
   placeholderKey?: string;
   helperTextKey?: string;
   options?: { value: string; labelKey: string }[];
   defaultValue?: string;
+  /**
+   * For `control: "connection"` - the connection-catalogue entry id this slot accepts (e.g.
+   * "sftp"). Filters the picker to matching connections and pins the inline "new connection" form.
+   */
+  connectionTypeId?: string;
 }
 
 /** A source type the wizard can create, with the fields its config needs. */
@@ -72,6 +89,64 @@ export interface CreatableSourceType {
   fields: SourceFieldDef[];
 }
 
+/**
+ * The config a network source (SFTP/FTP/SMB) needs: a stored connection of the matching protocol,
+ * the folder to poll, and the same consume/snapshot + recursion choices as a folder source. Shared
+ * copy across the three protocols, since only the connection type differs.
+ */
+function networkSourceFields(connectionTypeId: string): SourceFieldDef[] {
+  return [
+    {
+      key: "connectionId",
+      labelKey: "portal.sources.networkFields.connection.label",
+      control: "connection",
+      connectionTypeId,
+      required: true,
+      helperTextKey: "portal.sources.networkFields.connection.helperText",
+    },
+    {
+      key: "directory",
+      labelKey: "portal.sources.networkFields.directory.label",
+      control: "text",
+      placeholderKey: "portal.sources.networkFields.directory.placeholder",
+      helperTextKey: "portal.sources.networkFields.directory.helperText",
+    },
+    {
+      key: "mode",
+      labelKey: "portal.sources.networkFields.mode.label",
+      control: "select",
+      defaultValue: "consume",
+      helperTextKey: "portal.sources.networkFields.mode.helperText",
+      options: [
+        {
+          value: "consume",
+          labelKey: "portal.sources.networkFields.mode.options.consume",
+        },
+        {
+          value: "snapshot",
+          labelKey: "portal.sources.networkFields.mode.options.snapshot",
+        },
+      ],
+    },
+    {
+      key: "recursive",
+      labelKey: "portal.sources.networkFields.recursive.label",
+      control: "select",
+      defaultValue: "false",
+      options: [
+        {
+          value: "false",
+          labelKey: "portal.sources.networkFields.recursive.options.top",
+        },
+        {
+          value: "true",
+          labelKey: "portal.sources.networkFields.recursive.options.all",
+        },
+      ],
+    },
+  ];
+}
+
 export const CREATABLE_SOURCE_TYPES: CreatableSourceType[] = [
   {
     type: "folder",
@@ -182,6 +257,24 @@ export const CREATABLE_SOURCE_TYPES: CreatableSourceType[] = [
       },
     ],
   },
+  {
+    type: "sftp",
+    labelKey: "portal.sources.types.sftp.label",
+    descriptionKey: "portal.sources.types.sftp.description",
+    fields: networkSourceFields("sftp"),
+  },
+  {
+    type: "ftp",
+    labelKey: "portal.sources.types.ftp.label",
+    descriptionKey: "portal.sources.types.ftp.description",
+    fields: networkSourceFields("ftp"),
+  },
+  {
+    type: "network",
+    labelKey: "portal.sources.types.network.label",
+    descriptionKey: "portal.sources.types.network.description",
+    fields: networkSourceFields("smb"),
+  },
   {
     type: WEBHOOK_SOURCE_TYPE,
     labelKey: "portal.sources.types.webhook.label",
@@ -208,8 +301,6 @@ export const COMING_SOON_SOURCE_TYPES: ComingSoonSourceType[] = [
   "googledrive",
   "dropbox",
   "box",
-  "network",
-  "sftp",
   "email",
 ].map((type) => ({
   type,

gradle.properties

@@ -16,3 +16,8 @@ org.gradle.java.installations.auto-download=true
 
 org.gradle.daemon=true
 # org.gradle.configuration-cache=true
+
+# Gradle daemon heap. Without this the daemon uses Gradle's 512m default, which the SaaS build
+# variant (it compiles saas + proprietary + core together) exhausts during compileTestJava - the
+# GC thrashes and the daemon is stopped. Give all flavours comfortable headroom.
+org.gradle.jvmargs=-Xmx2g -XX:MaxMetaspaceSize=512m -Dfile.encoding=UTF-8