From a88531700cc9c695fdde3fd2b4c30fc862b4aa30 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jo=C3=A3o=20Magalh=C3=A3es?= <joamag@gmail.com>
Date: Wed, 2 Sep 2026 18:52:48 +0100
Subject: [PATCH 1/6] fix(file,ftp,tftp): stop following a link out of the root
 that is served

- A link under the root pointing outside it was followed and served, which
  a service on a reserved port turns into a read by a privileged process
- The HTTP file server also took a sibling whose name started like the root
- A flag keeps the old behaviour for a root that publishes through links
---
 CHANGELOG.md                    |  4 +-
 src/netius/extra/file.py        | 15 ++++++-
 src/netius/extra/file.pyi       |  2 +
 src/netius/servers/ftp.py       | 28 ++++++++++--
 src/netius/servers/ftp.pyi      | 19 +++++++-
 src/netius/servers/tftp.py      | 16 ++++++-
 src/netius/servers/tftp.pyi     |  5 ++-
 src/netius/test/extra/file.py   | 80 +++++++++++++++++++++++++++++++++
 src/netius/test/servers/ftp.py  | 51 +++++++++++++++++++++
 src/netius/test/servers/tftp.py | 62 +++++++++++++++++++++++--
 10 files changed, 267 insertions(+), 15 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 90e221e0..2fa606b8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-*
+* A file service can now be told to follow a link that leaves the root it serves - [#126](https://github.com/hivesolutions/netius/issues/126)
 
 ### Changed
 
@@ -19,11 +19,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 * A configuration file is now read under Python 3.5, instead of failing on it
 * A header that is set by its position now replaces the one that is there
+* A link that leaves the root of a file service is no longer followed by default - [#126](https://github.com/hivesolutions/netius/issues/126)
 * A string that carries no null byte is now given back whole instead of raising
 * A TFTP file that is served under an absolute name no longer fails to open
 * A TFTP peer can no longer reach a file outside the root that is served
 * An environment value that is unset can now be forced into a file, instead of raising
 * An FTP peer can no longer reach a file outside the root that is served
+* An HTTP file server can no longer reach a directory whose name only starts like the root - [#126](https://github.com/hivesolutions/netius/issues/126)
 * The allowed addresses of a service are now honoured, instead of every peer being refused
 * The asyncio compatibility of the event loop no longer fails under Python 3.6
 * The check for a test run no longer raises when a caller carries no source
diff --git a/src/netius/extra/file.py b/src/netius/extra/file.py
index 71f3d054..7ec11705 100644
--- a/src/netius/extra/file.py
+++ b/src/netius/extra/file.py
@@ -88,6 +88,7 @@ def __init__(
         list_engine="base",
         cors=False,
         cache=0,
+        follow_links=False,
         *args,
         **kwargs
     ):
@@ -100,6 +101,7 @@ def __init__(
         self.list_engine = list_engine
         self.cors = cors
         self.cache = cache
+        self.follow_links = follow_links
 
     @classmethod
     def _sorter_build(cls, name=None):
@@ -497,6 +499,10 @@ def on_serve(self):
             self.cors = self.get_env("CORS", self.cors, cast=bool)
         if self.env:
             self.cache = self.get_env("CACHE", self.cache, cast=int)
+        if self.env:
+            self.follow_links = self.get_env(
+                "FOLLOW_LINKS", self.follow_links, cast=bool
+            )
         self._build_regex()
         self.base_path = os.path.abspath(self.base_path)
         self.cache_d = datetime.timedelta(seconds=self.cache)
@@ -547,7 +553,14 @@ def on_data_http(self, connection, parser):
             # verifies if the provided path starts with the contents of the
             # base path in case it does not it's a security issue and a proper
             # exception must be raised indicating the issue
-            is_sub = path_f.startswith(self.base_path)
+            base_path_r = self.base_path
+            path_r = path_f
+            if not self.follow_links:
+                base_path_r = os.path.realpath(base_path_r)
+                path_r = os.path.realpath(path_f)
+            is_sub = path_r == base_path_r or path_r.startswith(
+                os.path.join(base_path_r, "")
+            )
             if not is_sub:
                 raise netius.SecurityError("Invalid path")
 
diff --git a/src/netius/extra/file.pyi b/src/netius/extra/file.pyi
index 046520e0..c5be212e 100644
--- a/src/netius/extra/file.pyi
+++ b/src/netius/extra/file.pyi
@@ -19,6 +19,7 @@ class FileServer(HTTP2Server):
     list_engine: str
     cors: bool
     cache: int
+    follow_links: bool
 
     def __init__(
         self,
@@ -30,6 +31,7 @@ class FileServer(HTTP2Server):
         list_engine: str = ...,
         cors: bool = ...,
         cache: int = ...,
+        follow_links: bool = ...,
         *args,
         **kwargs
     ): ...
diff --git a/src/netius/servers/ftp.py b/src/netius/servers/ftp.py
index 99d75b02..cf9d198a 100644
--- a/src/netius/servers/ftp.py
+++ b/src/netius/servers/ftp.py
@@ -96,12 +96,19 @@
 class FTPConnection(netius.Connection):
 
     def __init__(
-        self, base_path="", host="ftp.localhost", mode="ascii", *args, **kwargs
+        self,
+        base_path="",
+        host="ftp.localhost",
+        mode="ascii",
+        follow_links=False,
+        *args,
+        **kwargs
     ):
         netius.Connection.__init__(self, *args, **kwargs)
         self.parser = None
         self.base_path = os.path.abspath(base_path)
         self.host = host
+        self.follow_links = follow_links
         self.mode = mode
         self.data_server = None
         self.remaining = None
@@ -497,8 +504,13 @@ def _get_path(self, extra=None):
         # verifies if the resolved path starts with the contents of the
         # base path in case it does not it's a security issue and a proper
         # exception must be raised indicating the issue
-        is_sub = relative_path == self.base_path or relative_path.startswith(
-            os.path.join(self.base_path, "")
+        base_path_r = self.base_path
+        relative_path_r = relative_path
+        if not self.follow_links:
+            base_path_r = os.path.realpath(base_path_r)
+            relative_path_r = os.path.realpath(relative_path)
+        is_sub = relative_path_r == base_path_r or relative_path_r.startswith(
+            os.path.join(base_path_r, "")
         )
         if not is_sub:
             raise netius.SecurityError("Invalid path")
@@ -559,10 +571,13 @@ class FTPServer(netius.ContainerServer):
     :see: http://tools.ietf.org/html/rfc959
     """
 
-    def __init__(self, base_path="", auth_s="dummy", *args, **kwargs):
+    def __init__(
+        self, base_path="", auth_s="dummy", follow_links=False, *args, **kwargs
+    ):
         netius.ContainerServer.__init__(self, *args, **kwargs)
         self.base_path = base_path
         self.auth_s = auth_s
+        self.follow_links = follow_links
 
     def serve(self, host="ftp.localhost", port=21, *args, **kwargs):
         netius.ContainerServer.serve(self, port=port, *args, **kwargs)
@@ -584,6 +599,10 @@ def on_serve(self):
             self.host = self.get_env("FTP_HOST", self.host)
         if self.env:
             self.auth_s = self.get_env("FTP_AUTH", self.auth_s)
+        if self.env:
+            self.follow_links = self.get_env(
+                "FOLLOW_LINKS", self.follow_links, cast=bool
+            )
         self.auth = self.get_auth(self.auth_s)
         self.info(
             "Starting FTP server on '%s' using '%s' ..." % (self.host, self.auth_s)
@@ -600,6 +619,7 @@ def build_connection(self, socket, address, ssl=False):
             ssl=ssl,
             base_path=self.base_path,
             host=self.host,
+            follow_links=self.follow_links,
         )
 
     def on_line_ftp(self, connection, code, message):
diff --git a/src/netius/servers/ftp.pyi b/src/netius/servers/ftp.pyi
index 3ae85fac..8785f9cb 100644
--- a/src/netius/servers/ftp.pyi
+++ b/src/netius/servers/ftp.pyi
@@ -16,6 +16,7 @@ class FTPConnection(Connection):
     base_path: str
     host: str
     mode: str
+    follow_links: bool
     data_server: FTPDataServer | None
     remaining: str | None
     cwd: str
@@ -23,7 +24,13 @@ class FTPConnection(Connection):
     password: str
 
     def __init__(
-        self, base_path: str = ..., host: str = ..., mode: str = ..., *args, **kwargs
+        self,
+        base_path: str = ...,
+        host: str = ...,
+        mode: str = ...,
+        follow_links: bool = ...,
+        *args,
+        **kwargs
     ): ...
     def open(self, *args, **kwargs): ...
     def close(self, *args, **kwargs): ...
@@ -119,8 +126,16 @@ class FTPDataServer(StreamServer):
 class FTPServer(ContainerServer):
     base_path: str
     auth_s: str
+    follow_links: bool
 
-    def __init__(self, base_path: str = ..., auth_s: str = ..., *args, **kwargs): ...
+    def __init__(
+        self,
+        base_path: str = ...,
+        auth_s: str = ...,
+        follow_links: bool = ...,
+        *args,
+        **kwargs
+    ): ...
     def serve(  # type: ignore[override]
         self, host: str = ..., port: int = ..., *args, **kwargs
     ): ...
diff --git a/src/netius/servers/tftp.py b/src/netius/servers/tftp.py
index 366a0081..25b7e4a0 100644
--- a/src/netius/servers/tftp.py
+++ b/src/netius/servers/tftp.py
@@ -116,7 +116,14 @@ def _get_file(self, allow_absolute=False):
         # verifies if the resolved path starts with the contents of the
         # base path in case it does not it's a security issue and a proper
         # exception must be raised indicating the issue
-        is_sub = path == base_path or path.startswith(os.path.join(base_path, ""))
+        base_path_r = base_path
+        path_r = path
+        if not self.owner.follow_links:
+            base_path_r = os.path.realpath(base_path_r)
+            path_r = os.path.realpath(path)
+        is_sub = path_r == base_path_r or path_r.startswith(
+            os.path.join(base_path_r, "")
+        )
         if not is_sub:
             raise netius.SecurityError("Invalid path")
 
@@ -231,9 +238,10 @@ class TFTPServer(netius.DatagramServer):
 
     ALLOWED_OPERATIONS = (netius.common.RRQ_TFTP, netius.common.ACK_TFTP)
 
-    def __init__(self, base_path="", *args, **kwargs):
+    def __init__(self, base_path="", follow_links=False, *args, **kwargs):
         netius.DatagramServer.__init__(self, *args, **kwargs)
         self.base_path = base_path
+        self.follow_links = follow_links
         self.sessions = dict()
 
     def serve(self, port=69, *args, **kwargs):
@@ -259,6 +267,10 @@ def on_serve(self):
         netius.DatagramServer.on_serve(self)
         if self.env:
             self.base_path = self.get_env("BASE_PATH", self.base_path)
+        if self.env:
+            self.follow_links = self.get_env(
+                "FOLLOW_LINKS", self.follow_links, cast=bool
+            )
         self.info("Starting TFTP server ...")
         self.info(
             "Defining '%s' as the root of the file server ..." % (self.base_path or ".")
diff --git a/src/netius/servers/tftp.pyi b/src/netius/servers/tftp.pyi
index e5414117..cf4c369d 100644
--- a/src/netius/servers/tftp.pyi
+++ b/src/netius/servers/tftp.pyi
@@ -53,9 +53,12 @@ class TFTPRequest:
 class TFTPServer(DatagramServer):
     ALLOWED_OPERATIONS: tuple[int, ...]
     base_path: str
+    follow_links: bool
     sessions: dict[Any, TFTPSession]
 
-    def __init__(self, base_path: str = ..., *args, **kwargs): ...
+    def __init__(
+        self, base_path: str = ..., follow_links: bool = ..., *args, **kwargs
+    ): ...
     def serve(self, port: int = ..., *args, **kwargs): ...  # type: ignore[override]
     def on_data(self, address: Any, data: bytes): ...
     def on_serve(self): ...
diff --git a/src/netius/test/extra/file.py b/src/netius/test/extra/file.py
index 06ff02d6..dfe208a6 100644
--- a/src/netius/test/extra/file.py
+++ b/src/netius/test/extra/file.py
@@ -79,6 +79,7 @@ def test_init(self):
             self.assertEqual(server.list_engine, "base")
             self.assertEqual(server.cors, False)
             self.assertEqual(server.cache, 0)
+            self.assertEqual(server.follow_links, False)
         finally:
             server.cleanup()
 
@@ -92,6 +93,7 @@ def test_init_values(self):
             list_engine="apache",
             cors=True,
             cache=60,
+            follow_links=True,
         )
 
         try:
@@ -103,6 +105,7 @@ def test_init_values(self):
             self.assertEqual(server.list_engine, "apache")
             self.assertEqual(server.cors, True)
             self.assertEqual(server.cache, 60)
+            self.assertEqual(server.follow_links, True)
         finally:
             server.cleanup()
 
@@ -359,6 +362,7 @@ def test_on_serve_env(self):
             LIST_ENGINE="apache",
             CORS=True,
             CACHE=60,
+            FOLLOW_LINKS=True,
         )
 
         try:
@@ -376,6 +380,7 @@ def test_on_serve_env(self):
             self.assertEqual(server.list_engine, "apache")
             self.assertEqual(server.cors, True)
             self.assertEqual(server.cache, 60)
+            self.assertEqual(server.follow_links, True)
         finally:
             server.cleanup()
 
@@ -446,6 +451,73 @@ def test_on_data_http_escape(self):
         # security error is reported as an internal server error instead
         self.assertEqual(connection.send_response.call_args[1]["code"], 500)
 
+    def test_on_data_http_sibling(self):
+        if mock == None:
+            self.skipTest("Skipping test: mock unavailable")
+
+        parent = os.path.dirname(self.base_path)
+        sibling = os.path.basename(self.base_path) + "-backup"
+        os.mkdir(os.path.join(parent, sibling))
+        self._write(os.path.join(parent, sibling, "secret.txt"), b"secret")
+
+        connection = self._make_connection()
+
+        try:
+            self.server.on_data_http(
+                connection, self._make_parser("/../%s/secret.txt" % sibling)
+            )
+
+            # a directory whose name only starts like the base path is not
+            # under it, so reaching into it may never be served either
+            self.assertEqual(connection.send_response.call_args[1]["code"], 500)
+        finally:
+            shutil.rmtree(os.path.join(parent, sibling))
+
+    def test_on_data_http_link(self):
+        if mock == None:
+            self.skipTest("Skipping test: mock unavailable")
+
+        outside = tempfile.mkdtemp()
+        self._write(os.path.join(outside, "secret.txt"), b"secret")
+        self._link(
+            os.path.join(outside, "secret.txt"),
+            os.path.join(self.base_path, "link.txt"),
+        )
+
+        connection = self._make_connection()
+
+        try:
+            self.server.on_data_http(connection, self._make_parser("/link.txt"))
+
+            # a link that sits under the base path but points out of it is a
+            # way around the containment, so the target of it is what counts
+            self.assertEqual(connection.send_response.call_args[1]["code"], 500)
+
+            # the operator may have published through the link on purpose, in
+            # which case the server is told to follow it
+            self.server.follow_links = True
+            connection = self._make_connection()
+            self.server.on_data_http(connection, self._make_parser("/link.txt"))
+
+            self.assertNotEqual(connection.send_response.call_args[1]["code"], 500)
+        finally:
+            self.server.follow_links = False
+            shutil.rmtree(outside)
+
+    def test_on_data_http_link_inside(self):
+        if mock == None:
+            self.skipTest("Skipping test: mock unavailable")
+
+        self._link(self.hello_path, os.path.join(self.base_path, "link.txt"))
+
+        connection = self._make_connection()
+
+        self.server.on_data_http(connection, self._make_parser("/link.txt"))
+
+        # a link whose target stays under the base path leads nowhere it
+        # should not, so it is served as the file that it names
+        self.assertNotEqual(connection.send_response.call_args[1]["code"], 500)
+
     def test_on_data_http_queue(self):
         if mock == None:
             self.skipTest("Skipping test: mock unavailable")
@@ -899,6 +971,14 @@ def _make_items(self):
             ),
         ]
 
+    def _link(self, source, target):
+        # the creation of a link is reserved to a privileged user under
+        # some systems, in which case the case has nothing to run
+        try:
+            os.symlink(source, target)
+        except (AttributeError, NotImplementedError, OSError):
+            self.skipTest("Skipping test: symbolic links unavailable")
+
     def _make_connection(self):
         connection = mock.MagicMock()
         # removes the dynamic attributes that the server checks through
diff --git a/src/netius/test/servers/ftp.py b/src/netius/test/servers/ftp.py
index 692c8c34..8b1d8be6 100644
--- a/src/netius/test/servers/ftp.py
+++ b/src/netius/test/servers/ftp.py
@@ -55,6 +55,7 @@ def setUp(self):
         self.connection.owner = self.server
         self.connection.cwd = "/"
         self.connection.mode = "ascii"
+        self.connection.follow_links = False
         self.sent = []
         self.connection.send_ftp = lambda code, message="", **kwargs: self.sent.append(
             (code, message)
@@ -140,6 +141,7 @@ def setUp(self):
         self.connection.base_path = os.path.abspath(self.base)
         self.connection.cwd = "/"
         self.connection.mode = "ascii"
+        self.connection.follow_links = False
         self.connection.data_server = None
         self.connection.remaining = None
         self.sent = []
@@ -504,6 +506,55 @@ def test__get_path_sibling(self):
         finally:
             shutil.rmtree(os.path.join(parent, sibling))
 
+    def test__get_path_link(self):
+        outside = tempfile.mkdtemp()
+        secret = os.path.join(outside, "secret.txt")
+        file = open(secret, "wb")
+        try:
+            file.write(b"secret")
+        finally:
+            file.close()
+        self._link(secret, os.path.join(self.base, "link.txt"))
+
+        try:
+            # a link that sits under the root but points out of it is a way
+            # around the containment, so the target of it is what counts
+            self.assertRaises(
+                netius.SecurityError, self.connection._get_path, "link.txt"
+            )
+
+            # the operator may have published through the link on purpose, in
+            # which case the service is told to follow it
+            self.connection.follow_links = True
+            self.assertEqual(
+                self.connection._get_path("link.txt"),
+                os.path.join(self.connection.base_path, "link.txt"),
+            )
+        finally:
+            self.connection.follow_links = False
+            shutil.rmtree(outside)
+
+    def test__get_path_link_inside(self):
+        self._store("file.txt", b"contents")
+        self._link(
+            os.path.join(self.base, "file.txt"), os.path.join(self.base, "link.txt")
+        )
+
+        # a link whose target stays under the root leads nowhere it should not,
+        # so it resolves as the file that it names
+        self.assertEqual(
+            self.connection._get_path("link.txt"),
+            os.path.join(self.connection.base_path, "link.txt"),
+        )
+
+    def _link(self, source, target):
+        # the creation of a link is reserved to a privileged user under
+        # some systems, in which case the case has nothing to run
+        try:
+            os.symlink(source, target)
+        except (AttributeError, NotImplementedError, OSError):
+            self.skipTest("Skipping test: symbolic links unavailable")
+
     def _store(self, name, contents):
         path = os.path.join(self.base, name)
         file = open(path, "wb")
diff --git a/src/netius/test/servers/tftp.py b/src/netius/test/servers/tftp.py
index 9f9ba73b..b889f00d 100644
--- a/src/netius/test/servers/tftp.py
+++ b/src/netius/test/servers/tftp.py
@@ -246,6 +246,51 @@ def test__get_file_sibling(self):
         finally:
             shutil.rmtree(os.path.join(parent, sibling))
 
+    def test__get_file_link(self):
+        outside = tempfile.mkdtemp()
+        secret = os.path.join(outside, "secret.txt")
+        file = open(secret, "wb")
+        try:
+            file.write(b"secret")
+        finally:
+            file.close()
+        self._link(secret, os.path.join(self.base, "link.txt"))
+
+        session = self._make_session("link.txt")
+
+        try:
+            # a link that sits under the root but points out of it is a way
+            # around the containment, so the target of it is what counts
+            self.assertRaises(netius.SecurityError, session._get_file)
+
+            # the operator may have published through the link on purpose, in
+            # which case the service is told to follow it
+            self.server.follow_links = True
+            self.assertEqual(session._get_file().read(), b"secret")
+        finally:
+            self.server.follow_links = False
+            shutil.rmtree(outside)
+
+    def test__get_file_link_inside(self):
+        self._store("file.txt", b"contents")
+        self._link(
+            os.path.join(self.base, "file.txt"), os.path.join(self.base, "link.txt")
+        )
+
+        session = self._make_session("link.txt")
+
+        # a link whose target stays under the root leads nowhere it should not,
+        # so it is served as the file that it names
+        self.assertEqual(session._get_file().read(), b"contents")
+
+    def _link(self, source, target):
+        # the creation of a link is reserved to a privileged user under
+        # some systems, in which case the case has nothing to run
+        try:
+            os.symlink(source, target)
+        except (AttributeError, NotImplementedError, OSError):
+            self.skipTest("Skipping test: symbolic links unavailable")
+
     def _make_session(self, name):
         # builds a session of the service and keeps it, so that the file it
         # reads is released once the case is over
@@ -455,16 +500,25 @@ def test_on_data_error(self):
         self.assertEqual(struct.unpack("!H", data[:2])[0], netius.common.ERROR_TFTP)
 
     def test_on_serve(self):
+        values = dict(BASE_PATH="/other", FOLLOW_LINKS=True)
+
         with mock.patch.object(
-            self.server, "get_env", return_value="/other"
+            self.server,
+            "get_env",
+            side_effect=lambda name, default, **kwargs: values.get(name, default),
         ) as get_env:
             self.server.env = True
             self.server.on_serve()
 
-        # with an environment to read from, the root of the file service is
-        # the one that it names
-        self.assertEqual(get_env.call_args[0][0], "BASE_PATH")
+        # with an environment to read from, both the root of the file service
+        # and whether a link is followed out of it are the ones that it names
+        names = [call[0][0] for call in get_env.call_args_list]
+        self.assertEqual("BASE_PATH" in names, True)
+        self.assertEqual("FOLLOW_LINKS" in names, True)
         self.assertEqual(self.server.base_path, "/other")
+        self.assertEqual(self.server.follow_links, True)
+
+        self.server.follow_links = False
 
         self.server.env = False
         self.server.base_path = self.base

From 79266eec122431fc85c1e7905436da68bbe6b6a0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jo=C3=A3o=20Magalh=C3=A3es?= <joamag@gmail.com>
Date: Wed, 2 Sep 2026 18:58:09 +0100
Subject: [PATCH 2/6] test: release the file of a link case before the
 directory is removed

- The service keeps the file that it opened, which a system that locks an
  open one will not let go of while the directory is taken away
---
 src/netius/test/extra/file.py   | 1 +
 src/netius/test/servers/tftp.py | 1 +
 2 files changed, 2 insertions(+)

diff --git a/src/netius/test/extra/file.py b/src/netius/test/extra/file.py
index dfe208a6..3673cab9 100644
--- a/src/netius/test/extra/file.py
+++ b/src/netius/test/extra/file.py
@@ -502,6 +502,7 @@ def test_on_data_http_link(self):
             self.assertNotEqual(connection.send_response.call_args[1]["code"], 500)
         finally:
             self.server.follow_links = False
+            self._close_files()
             shutil.rmtree(outside)
 
     def test_on_data_http_link_inside(self):
diff --git a/src/netius/test/servers/tftp.py b/src/netius/test/servers/tftp.py
index b889f00d..2f02a080 100644
--- a/src/netius/test/servers/tftp.py
+++ b/src/netius/test/servers/tftp.py
@@ -269,6 +269,7 @@ def test__get_file_link(self):
             self.assertEqual(session._get_file().read(), b"secret")
         finally:
             self.server.follow_links = False
+            session.close()
             shutil.rmtree(outside)
 
     def test__get_file_link_inside(self):

From 816267467b1a400b6214e778bf86465e5dd91ee3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jo=C3=A3o=20Magalh=C3=A3es?= <joamag@gmail.com>
Date: Wed, 2 Sep 2026 19:15:26 +0100
Subject: [PATCH 3/6] fix(ftp): answer a refused path instead of dropping the
 connection

- A client that walks up from the root sends a command that is ordinary,
  so the refusal of it belongs in the protocol rather than in a close
- The option that follows a link is now documented with the others
- A case that skips no longer leaves the root it made behind
---
 doc/configuration.md            |  1 +
 src/netius/servers/ftp.py       |  9 +++++++--
 src/netius/test/extra/file.py   | 15 ++++++++-------
 src/netius/test/servers/ftp.py  | 24 +++++++++++++++++-------
 src/netius/test/servers/tftp.py | 22 ++++++++++++----------
 5 files changed, 45 insertions(+), 26 deletions(-)

diff --git a/doc/configuration.md b/doc/configuration.md
index 3e8a8c0d..7deaca1d 100644
--- a/doc/configuration.md
+++ b/doc/configuration.md
@@ -74,6 +74,7 @@
 | **PATH_REGEX**  | `list` | `[]`    | The list of regex to path values (separated by the `:` character) that provide a simple way of URL re-writing like behaviour under the file serving extension (eg: `.*:index.html`). |
 | **LIST_DIRS**   | `bool` | `True`  | If directory listing is enabled (may pose a security issue).                                                                                                                         |
 | **LIST_ENGINE** | `str`  | `base`  | The name of the HTML generation engine to be used while listing files (eg: base, apache, legacy, etc.).                                                                              |
+| **FOLLOW_LINKS** | `bool` | `False` | If a symbolic link under the base path whose target lies outside of it should be followed, by default such a link is refused (may pose a security issue).                            |
 
 #### HTTP
 
diff --git a/src/netius/servers/ftp.py b/src/netius/servers/ftp.py
index cf9d198a..a886b3f7 100644
--- a/src/netius/servers/ftp.py
+++ b/src/netius/servers/ftp.py
@@ -265,9 +265,14 @@ def on_line(self, code, message, is_final=True):
 
         # retrieves the reference to the method that is going to be called
         # for the handling of the current line from the current instance and
-        # then calls it with the provided message
+        # then calls it with the provided message, note that a path that is
+        # refused is reported back as a failed command instead of being
+        # raised, as dropping the connection for it would be too much
         method = getattr(self, method_n)
-        method(message)
+        try:
+            method(message)
+        except netius.SecurityError:
+            self.not_ok()
 
     def on_user(self, message):
         self.username = message
diff --git a/src/netius/test/extra/file.py b/src/netius/test/extra/file.py
index 3673cab9..6d0964af 100644
--- a/src/netius/test/extra/file.py
+++ b/src/netius/test/extra/file.py
@@ -478,15 +478,16 @@ def test_on_data_http_link(self):
             self.skipTest("Skipping test: mock unavailable")
 
         outside = tempfile.mkdtemp()
-        self._write(os.path.join(outside, "secret.txt"), b"secret")
-        self._link(
-            os.path.join(outside, "secret.txt"),
-            os.path.join(self.base_path, "link.txt"),
-        )
-
-        connection = self._make_connection()
 
         try:
+            self._write(os.path.join(outside, "secret.txt"), b"secret")
+            self._link(
+                os.path.join(outside, "secret.txt"),
+                os.path.join(self.base_path, "link.txt"),
+            )
+
+            connection = self._make_connection()
+
             self.server.on_data_http(connection, self._make_parser("/link.txt"))
 
             # a link that sits under the base path but points out of it is a
diff --git a/src/netius/test/servers/ftp.py b/src/netius/test/servers/ftp.py
index 8b1d8be6..09d935a6 100644
--- a/src/netius/test/servers/ftp.py
+++ b/src/netius/test/servers/ftp.py
@@ -232,6 +232,15 @@ def test_data_ftp(self):
         finally:
             file.close()
 
+    def test_on_line_refused(self):
+        self.connection.on_line("CWD", "..")
+
+        # a path that walks out of the root is a command that failed rather
+        # than a reason to drop the connection of the peer, which is what an
+        # ordinary client asks for when it walks up from the root
+        self.assertEqual(self.sent[-1], (500, "not ok"))
+        self.assertEqual(self.connection.cwd, "/")
+
     def test_on_syst(self):
         self.connection.on_syst("")
 
@@ -508,15 +517,16 @@ def test__get_path_sibling(self):
 
     def test__get_path_link(self):
         outside = tempfile.mkdtemp()
-        secret = os.path.join(outside, "secret.txt")
-        file = open(secret, "wb")
-        try:
-            file.write(b"secret")
-        finally:
-            file.close()
-        self._link(secret, os.path.join(self.base, "link.txt"))
 
         try:
+            secret = os.path.join(outside, "secret.txt")
+            file = open(secret, "wb")
+            try:
+                file.write(b"secret")
+            finally:
+                file.close()
+            self._link(secret, os.path.join(self.base, "link.txt"))
+
             # a link that sits under the root but points out of it is a way
             # around the containment, so the target of it is what counts
             self.assertRaises(
diff --git a/src/netius/test/servers/tftp.py b/src/netius/test/servers/tftp.py
index 2f02a080..3966d2f7 100644
--- a/src/netius/test/servers/tftp.py
+++ b/src/netius/test/servers/tftp.py
@@ -248,17 +248,18 @@ def test__get_file_sibling(self):
 
     def test__get_file_link(self):
         outside = tempfile.mkdtemp()
-        secret = os.path.join(outside, "secret.txt")
-        file = open(secret, "wb")
-        try:
-            file.write(b"secret")
-        finally:
-            file.close()
-        self._link(secret, os.path.join(self.base, "link.txt"))
-
-        session = self._make_session("link.txt")
+        session = None
 
         try:
+            secret = os.path.join(outside, "secret.txt")
+            file = open(secret, "wb")
+            try:
+                file.write(b"secret")
+            finally:
+                file.close()
+            self._link(secret, os.path.join(self.base, "link.txt"))
+
+            session = self._make_session("link.txt")
             # a link that sits under the root but points out of it is a way
             # around the containment, so the target of it is what counts
             self.assertRaises(netius.SecurityError, session._get_file)
@@ -269,7 +270,8 @@ def test__get_file_link(self):
             self.assertEqual(session._get_file().read(), b"secret")
         finally:
             self.server.follow_links = False
-            session.close()
+            if session:
+                session.close()
             shutil.rmtree(outside)
 
     def test__get_file_link_inside(self):

From afd0f8c801e84e7c89be8d38da484449c9f68ece Mon Sep 17 00:00:00 2001
From: Joao Magalhaes <joamag@gmail.com>
Date: Thu, 3 Sep 2026 17:26:21 +0100
Subject: [PATCH 4/6] fix(ftp): refuse a transfer at the command instead of at
 the flushing

- The path of a file that is read or stored was only resolved once the data
  connection flushed, which may run after it is accepted, so a refusal there
  was raised past the handler with the transfer already announced
- The path is now resolved and kept at the command itself, where a refusal
  is answered as a failed command before anything is left over
---
 src/netius/servers/ftp.py      | 18 +++++++-----
 src/netius/test/servers/ftp.py | 53 ++++++++++++++++++++++++++++++----
 2 files changed, 58 insertions(+), 13 deletions(-)

diff --git a/src/netius/servers/ftp.py b/src/netius/servers/ftp.py
index a886b3f7..b9a02877 100644
--- a/src/netius/servers/ftp.py
+++ b/src/netius/servers/ftp.py
@@ -221,16 +221,13 @@ def flush_list(self):
 
     def flush_retr(self):
         self.send_ftp(150, "file sending")
-        full_path = self._get_path(self.file_name)
-        self.bytes_p = os.path.getsize(full_path)
-        self.file = open(full_path, "rb")
+        self.bytes_p = os.path.getsize(self.file_path)
+        self.file = open(self.file_path, "rb")
         self._file_send(self)
 
     def flush_stor(self):
         self.send_ftp(150, "file receiving")
-
-        full_path = self._get_path(self.file_name)
-        self.file = open(full_path, "wb")
+        self.file = open(self.file_path, "wb")
 
     def on_flush_list(self, connection):
         self._data_close()
@@ -399,13 +396,18 @@ def on_list(self, message):
         self.data_server.flush_ftp()
 
     def on_retr(self, message):
+        # resolves the path of the file at the command itself, so that a
+        # path that is refused is answered as a failed command before the
+        # transfer is announced, note that the flushing of the transfer may
+        # only run once the data connection is accepted, at which point the
+        # refusal would have no command left to be reported against
+        self.file_path = self._get_path(extra=message)
         self.remaining = "retr"
-        self.file_name = message
         self.data_server.flush_ftp()
 
     def on_stor(self, message):
+        self.file_path = self._get_path(extra=message)
         self.remaining = "stor"
-        self.file_name = message
         self.data_server.flush_ftp()
 
     def _file_send(self, connection):
diff --git a/src/netius/test/servers/ftp.py b/src/netius/test/servers/ftp.py
index 09d935a6..b2451206 100644
--- a/src/netius/test/servers/ftp.py
+++ b/src/netius/test/servers/ftp.py
@@ -241,6 +241,25 @@ def test_on_line_refused(self):
         self.assertEqual(self.sent[-1], (500, "not ok"))
         self.assertEqual(self.connection.cwd, "/")
 
+    def test_on_line_refused_retr(self):
+        self.connection.data_server = mock.MagicMock()
+
+        self.connection.on_line("RETR", "../secret.txt")
+
+        # a transfer whose path walks out of the root is refused at the
+        # command itself, before it is announced and before anything is left
+        # over for the data connection, which may only be accepted later on
+        # and would then have no command left to report the refusal against
+        self.assertEqual(self.sent, [(500, "not ok")])
+        self.assertEqual(self.connection.remaining, None)
+        self.assertEqual(self.connection.data_server.flush_ftp.called, False)
+
+        self.connection.on_line("STOR", "../secret.txt")
+
+        self.assertEqual(self.sent[-1], (500, "not ok"))
+        self.assertEqual(self.connection.remaining, None)
+        self.assertEqual(self.connection.data_server.flush_ftp.called, False)
+
     def test_on_syst(self):
         self.connection.on_syst("")
 
@@ -414,18 +433,42 @@ def test_on_list(self):
 
         self.connection.on_retr("file.txt")
 
-        # the same holds for a file that is read, the name of it being kept
-        # for the flushing that follows
+        # the same holds for a file that is read, the path of it being
+        # resolved at the command and kept for the flushing that follows
         self.assertEqual(self.connection.remaining, "retr")
-        self.assertEqual(self.connection.file_name, "file.txt")
+        self.assertEqual(
+            self.connection.file_path,
+            os.path.join(self.connection.base_path, "file.txt"),
+        )
 
         self.connection.on_stor("other.txt")
 
         self.assertEqual(self.connection.remaining, "stor")
-        self.assertEqual(self.connection.file_name, "other.txt")
+        self.assertEqual(
+            self.connection.file_path,
+            os.path.join(self.connection.base_path, "other.txt"),
+        )
+
+    def test_flush_retr(self):
+        self._store("file.txt", b"contents")
+        self.connection.data_server = mock.MagicMock()
+        self.connection.file_path = os.path.join(self.base, "file.txt")
+
+        self.connection.flush_retr()
+        try:
+            # the file whose path was resolved at the command is the one that
+            # is opened and sent over the data connection, the peer being told
+            # that the transfer starts
+            self.assertEqual(self.sent[-1][0], 150)
+            self.assertEqual(self.connection.file.closed, False)
+            self.assertEqual(
+                self.connection.data_server.send_ftp.call_args[0][0], b"contents"
+            )
+        finally:
+            self.connection.file.close()
 
     def test_flush_stor(self):
-        self.connection.file_name = "file.txt"
+        self.connection.file_path = os.path.join(self.base, "file.txt")
 
         self.connection.flush_stor()
         try:

From 126d05cc7f847e17b1da45148690ab8d6d811e10 Mon Sep 17 00:00:00 2001
From: Joao Magalhaes <joamag@gmail.com>
Date: Thu, 3 Sep 2026 17:26:21 +0100
Subject: [PATCH 5/6] fix(file): contain the index file that a directory is
 served through

- The containment only checked the directory that was asked for, while the
  index file taken from it is what is opened, so a link there that leaves
  the root was still followed and served
- The check is now a helper of its own, run for the index file as well
- The cases of a link that is served assert the file rather than no failure
---
 src/netius/extra/file.py      | 27 +++++++---
 src/netius/extra/file.pyi     |  1 +
 src/netius/test/extra/file.py | 98 ++++++++++++++++++++++++++++++++++-
 3 files changed, 116 insertions(+), 10 deletions(-)

diff --git a/src/netius/extra/file.py b/src/netius/extra/file.py
index 7ec11705..1a9574eb 100644
--- a/src/netius/extra/file.py
+++ b/src/netius/extra/file.py
@@ -553,14 +553,7 @@ def on_data_http(self, connection, parser):
             # verifies if the provided path starts with the contents of the
             # base path in case it does not it's a security issue and a proper
             # exception must be raised indicating the issue
-            base_path_r = self.base_path
-            path_r = path_f
-            if not self.follow_links:
-                base_path_r = os.path.realpath(base_path_r)
-                path_r = os.path.realpath(path_f)
-            is_sub = path_r == base_path_r or path_r.startswith(
-                os.path.join(base_path_r, "")
-            )
+            is_sub = self._is_sub(path_f)
             if not is_sub:
                 raise netius.SecurityError("Invalid path")
 
@@ -608,6 +601,13 @@ def on_dir_file(self, connection, parser, path, style=True):
             index_path = os.path.join(path, index_file)
             if not os.path.exists(index_path):
                 continue
+
+            # verifies that the index file is itself under the base path, as
+            # it's the one that is going to be opened and it may be a link
+            # whose target lies outside of it (security issue)
+            is_sub = self._is_sub(index_path)
+            if not is_sub:
+                raise netius.SecurityError("Invalid path")
             return self.on_normal_file(connection, parser, index_path)
 
         if not self.list_dirs:
@@ -850,6 +850,17 @@ def _resolve_regex(self, path):
             return (value, True)
         return (path, False)
 
+    def _is_sub(self, path):
+        # resolves both the base path and the provided path, unless a link
+        # is meant to be followed, so that a link under the base path whose
+        # target lies outside of it is not taken as being under it, note that
+        # the base path is also resolved as it may be reached through a link
+        base_path = self.base_path
+        if not self.follow_links:
+            base_path = os.path.realpath(base_path)
+            path = os.path.realpath(path)
+        return path == base_path or path.startswith(os.path.join(base_path, ""))
+
 
 if __name__ == "__main__":
     import logging
diff --git a/src/netius/extra/file.pyi b/src/netius/extra/file.pyi
index c5be212e..4f462eae 100644
--- a/src/netius/extra/file.pyi
+++ b/src/netius/extra/file.pyi
@@ -108,3 +108,4 @@ class FileServer(HTTP2Server):
     def _resolve(self, path: str) -> str: ...
     def _build_regex(self): ...
     def _resolve_regex(self, path: str) -> tuple[str, bool]: ...
+    def _is_sub(self, path: str) -> bool: ...
diff --git a/src/netius/test/extra/file.py b/src/netius/test/extra/file.py
index 6d0964af..1b0e2a5c 100644
--- a/src/netius/test/extra/file.py
+++ b/src/netius/test/extra/file.py
@@ -500,7 +500,7 @@ def test_on_data_http_link(self):
             connection = self._make_connection()
             self.server.on_data_http(connection, self._make_parser("/link.txt"))
 
-            self.assertNotEqual(connection.send_response.call_args[1]["code"], 500)
+            self.assertEqual(connection.send_response.call_args[1]["code"], 200)
         finally:
             self.server.follow_links = False
             self._close_files()
@@ -518,7 +518,10 @@ def test_on_data_http_link_inside(self):
 
         # a link whose target stays under the base path leads nowhere it
         # should not, so it is served as the file that it names
-        self.assertNotEqual(connection.send_response.call_args[1]["code"], 500)
+        self.assertEqual(connection.send_response.call_args[1]["code"], 200)
+        self.assertEqual(
+            connection.send_response.call_args[1]["headers"]["content-length"], "11"
+        )
 
     def test_on_data_http_queue(self):
         if mock == None:
@@ -599,6 +602,62 @@ def test_on_dir_file_index_missing(self):
             connection.send_response.call_args[1]["headers"]["content-length"], "12"
         )
 
+    def test_on_dir_file_index_link(self):
+        if mock == None:
+            self.skipTest("Skipping test: mock unavailable")
+
+        outside = tempfile.mkdtemp()
+
+        try:
+            self._write(os.path.join(outside, "secret.html"), b"<p>secret</p>")
+            self._link(
+                os.path.join(outside, "secret.html"),
+                os.path.join(self.base_path, "sub", "index.html"),
+            )
+            self.server.index_files = ["index.html"]
+
+            connection = self._make_connection()
+
+            self.server.on_data_http(connection, self._make_parser("/sub/"))
+
+            # the index file is the one that is opened, so one that is a link
+            # out of the base path is refused as the directory itself would be
+            self.assertEqual(connection.send_response.call_args[1]["code"], 500)
+
+            # the operator may have published through the link on purpose, in
+            # which case the server is told to follow it
+            self.server.follow_links = True
+            connection = self._make_connection()
+            self.server.on_data_http(connection, self._make_parser("/sub/"))
+
+            self.assertEqual(connection.send_response.call_args[1]["code"], 200)
+            self.assertEqual(
+                connection.send_response.call_args[1]["headers"]["content-length"],
+                "13",
+            )
+        finally:
+            self.server.follow_links = False
+            self._close_files()
+            shutil.rmtree(outside)
+
+    def test_on_dir_file_index_link_inside(self):
+        if mock == None:
+            self.skipTest("Skipping test: mock unavailable")
+
+        self._link(self.hello_path, os.path.join(self.base_path, "sub", "index.html"))
+        self.server.index_files = ["index.html"]
+
+        connection = self._make_connection()
+
+        self.server.on_data_http(connection, self._make_parser("/sub/"))
+
+        # an index file that links to a file under the base path leads nowhere
+        # it should not, so it is served as the file that it names
+        self.assertEqual(connection.send_response.call_args[1]["code"], 200)
+        self.assertEqual(
+            connection.send_response.call_args[1]["headers"]["content-length"], "11"
+        )
+
     def test_on_dir_file_no_list(self):
         if mock == None:
             self.skipTest("Skipping test: mock unavailable")
@@ -941,6 +1000,41 @@ def test_resolve_regex(self):
             self.server._resolve_regex("assets/logo.png"), ("assets/logo.png", False)
         )
 
+    def test_is_sub(self):
+        # the base path itself and anything under it are contained, while
+        # the parent and a sibling whose name only starts like it are not
+        self.assertEqual(self.server._is_sub(self.server.base_path), True)
+        self.assertEqual(self.server._is_sub(self.hello_path), True)
+        self.assertEqual(
+            self.server._is_sub(os.path.dirname(self.server.base_path)), False
+        )
+        self.assertEqual(self.server._is_sub(self.server.base_path + "-backup"), False)
+
+    def test_is_sub_link(self):
+        outside = tempfile.mkdtemp()
+
+        try:
+            self._write(os.path.join(outside, "secret.txt"), b"secret")
+            self._link(
+                os.path.join(outside, "secret.txt"),
+                os.path.join(self.base_path, "link.txt"),
+            )
+
+            # a link is judged by the target of it, unless the server is told
+            # to follow it, in which case the name under the base path is enough
+            self.assertEqual(
+                self.server._is_sub(os.path.join(self.base_path, "link.txt")), False
+            )
+
+            self.server.follow_links = True
+
+            self.assertEqual(
+                self.server._is_sub(os.path.join(self.base_path, "link.txt")), True
+            )
+        finally:
+            self.server.follow_links = False
+            shutil.rmtree(outside)
+
     def _make_items(self):
         return [
             dict(

From b3692cf98ec6111905257bfa1363cf6184656af5 Mon Sep 17 00:00:00 2001
From: Joao Magalhaes <joamag@gmail.com>
Date: Thu, 3 Sep 2026 17:45:50 +0100
Subject: [PATCH 6/6] perf(common): resolve a path only when a link sits below
 the root

- The containment resolved both the root and the path from the top of the
  file system on every request, while only a link below the root can lead
  out of it and what sits above resolves the same for both of them
- The components below the root are now checked with a stat each and the
  resolution only happens when one of them is a link
---
 src/netius/common/__init__.py  |   1 +
 src/netius/common/util.py      |  64 +++++++++++++
 src/netius/common/util.pyi     |   2 +
 src/netius/extra/file.py       |  12 +--
 src/netius/servers/ftp.py      |   9 +-
 src/netius/servers/tftp.py     |   9 +-
 src/netius/test/common/util.py | 164 +++++++++++++++++++++++++++++++++
 7 files changed, 238 insertions(+), 23 deletions(-)

diff --git a/src/netius/common/__init__.py b/src/netius/common/__init__.py
index e1f4f11c..e3d6cba4 100644
--- a/src/netius/common/__init__.py
+++ b/src/netius/common/__init__.py
@@ -200,6 +200,7 @@
     random_integer,
     host,
     hostname,
+    is_sub_path,
     size_round_unit,
     verify,
     verify_equal,
diff --git a/src/netius/common/util.py b/src/netius/common/util.py
index 1e23cc62..02d7046f 100644
--- a/src/netius/common/util.py
+++ b/src/netius/common/util.py
@@ -38,6 +38,7 @@
 """ The license for the module """
 
 import os
+import stat
 import math
 import socket
 import collections
@@ -66,6 +67,11 @@
 to be used for the string representation in the round
 based conversion of size units to be performed """
 
+FILE_ATTRIBUTE_REPARSE_POINT = 0x400
+""" The attribute that marks a file as a reparse point under
+Windows, the kind of a junction, which the mode of the file does
+not report as a link and yet leads somewhere else in the system """
+
 _HOST = None
 """ The globally cached value for the current hostname,
 this value is used to avoid an excessive blocking in the
@@ -297,6 +303,64 @@ def hostname():
     return socket.gethostname()
 
 
+def is_sub_path(base_path, path, follow_links=False):
+    """
+    Verifies if the provided path is contained by the base path,
+    meaning that it's either the base path itself or that it sits
+    below it, both of the paths are expected to be absolute and
+    normalized (no relative components).
+
+    Unless links are meant to be followed a link that sits below
+    the base path is resolved, so that one whose target lies outside
+    of it is not taken as contained, what sits above the base path
+    is never resolved as it's the same for both of the paths.
+
+    :type base_path: String
+    :param base_path: The path of the root that is expected to
+    contain the path that is going to be verified.
+    :type path: String
+    :param path: The path that is going to be verified for the
+    containment under the base path.
+    :type follow_links: bool
+    :param follow_links: If a link that sits below the base path
+    should be taken by its name rather than by its target.
+    :rtype: bool
+    :return: If the path is contained by the base path.
+    """
+
+    is_sub = path == base_path or path.startswith(os.path.join(base_path, ""))
+    if not is_sub or follow_links:
+        return is_sub
+
+    # walks the components of the path that sit below the base path
+    # looking for a link, in case none is found the path leads to the
+    # same place as the one it names and nothing else is to be verified,
+    # note that a component that is not there (or cannot be read) has
+    # nothing left to be resolved and so the walk ends at it
+    current = base_path
+    for name in path[len(base_path) :].split(os.sep):
+        if not name:
+            continue
+        current = os.path.join(current, name)
+        try:
+            mode = os.lstat(current)
+        except OSError:
+            return True
+        attributes = getattr(mode, "st_file_attributes", 0)
+        is_link = stat.S_ISLNK(mode.st_mode)
+        is_link |= bool(attributes & FILE_ATTRIBUTE_REPARSE_POINT)
+        if is_link:
+            break
+    else:
+        return True
+
+    # a link was found below the base path, so both of the paths are
+    # resolved and the containment is verified against their targets
+    base_path_r = os.path.realpath(base_path)
+    path_r = os.path.realpath(path)
+    return path_r == base_path_r or path_r.startswith(os.path.join(base_path_r, ""))
+
+
 def size_round_unit(
     size_value,
     minimum=DEFAULT_MINIMUM,
diff --git a/src/netius/common/util.pyi b/src/netius/common/util.pyi
index 59409b1f..9626555a 100644
--- a/src/netius/common/util.pyi
+++ b/src/netius/common/util.pyi
@@ -5,6 +5,7 @@ SIZE_UNITS_LIST_S: tuple[str, ...]
 SIZE_UNIT_COEFFICIENT: int
 DEFAULT_MINIMUM: int
 DEFAULT_PLACES: int
+FILE_ATTRIBUTE_REPARSE_POINT: int
 
 def cstring(value: str) -> str: ...
 def chunks(
@@ -27,6 +28,7 @@ def bytes_to_integer(bytes: bytes) -> int: ...
 def random_integer(number_bits: int) -> int: ...
 def host(default: str = ...) -> str: ...
 def hostname() -> str: ...
+def is_sub_path(base_path: str, path: str, follow_links: bool = ...) -> bool: ...
 def size_round_unit(
     size_value: int | float,
     minimum: int = ...,
diff --git a/src/netius/extra/file.py b/src/netius/extra/file.py
index 1a9574eb..cd70e189 100644
--- a/src/netius/extra/file.py
+++ b/src/netius/extra/file.py
@@ -851,15 +851,9 @@ def _resolve_regex(self, path):
         return (path, False)
 
     def _is_sub(self, path):
-        # resolves both the base path and the provided path, unless a link
-        # is meant to be followed, so that a link under the base path whose
-        # target lies outside of it is not taken as being under it, note that
-        # the base path is also resolved as it may be reached through a link
-        base_path = self.base_path
-        if not self.follow_links:
-            base_path = os.path.realpath(base_path)
-            path = os.path.realpath(path)
-        return path == base_path or path.startswith(os.path.join(base_path, ""))
+        return netius.common.is_sub_path(
+            self.base_path, path, follow_links=self.follow_links
+        )
 
 
 if __name__ == "__main__":
diff --git a/src/netius/servers/ftp.py b/src/netius/servers/ftp.py
index b9a02877..c034c1bf 100644
--- a/src/netius/servers/ftp.py
+++ b/src/netius/servers/ftp.py
@@ -511,13 +511,8 @@ def _get_path(self, extra=None):
         # verifies if the resolved path starts with the contents of the
         # base path in case it does not it's a security issue and a proper
         # exception must be raised indicating the issue
-        base_path_r = self.base_path
-        relative_path_r = relative_path
-        if not self.follow_links:
-            base_path_r = os.path.realpath(base_path_r)
-            relative_path_r = os.path.realpath(relative_path)
-        is_sub = relative_path_r == base_path_r or relative_path_r.startswith(
-            os.path.join(base_path_r, "")
+        is_sub = netius.common.is_sub_path(
+            self.base_path, relative_path, follow_links=self.follow_links
         )
         if not is_sub:
             raise netius.SecurityError("Invalid path")
diff --git a/src/netius/servers/tftp.py b/src/netius/servers/tftp.py
index 25b7e4a0..307fa19a 100644
--- a/src/netius/servers/tftp.py
+++ b/src/netius/servers/tftp.py
@@ -116,13 +116,8 @@ def _get_file(self, allow_absolute=False):
         # verifies if the resolved path starts with the contents of the
         # base path in case it does not it's a security issue and a proper
         # exception must be raised indicating the issue
-        base_path_r = base_path
-        path_r = path
-        if not self.owner.follow_links:
-            base_path_r = os.path.realpath(base_path_r)
-            path_r = os.path.realpath(path)
-        is_sub = path_r == base_path_r or path_r.startswith(
-            os.path.join(base_path_r, "")
+        is_sub = netius.common.is_sub_path(
+            base_path, path, follow_links=self.owner.follow_links
         )
         if not is_sub:
             raise netius.SecurityError("Invalid path")
diff --git a/src/netius/test/common/util.py b/src/netius/test/common/util.py
index 1bcaf559..76b301ea 100644
--- a/src/netius/test/common/util.py
+++ b/src/netius/test/common/util.py
@@ -28,6 +28,9 @@
 __license__ = "Apache License, Version 2.0"
 """ The license for the module """
 
+import os
+import shutil
+import tempfile
 import unittest
 
 import netius.common
@@ -167,6 +170,152 @@ def test_hostname(self):
         self.assertEqual(netius.legacy.is_string(result), True)
         self.assertEqual(len(result) > 0, True)
 
+    def test_is_sub_path(self):
+        base = os.path.abspath(os.path.join(os.sep, "srv", "ftp"))
+
+        # the base path itself and anything below it are contained, the
+        # separator being what tells a child apart from a sibling whose
+        # name only starts like the base path
+        self.assertEqual(netius.common.is_sub_path(base, base), True)
+        self.assertEqual(
+            netius.common.is_sub_path(base, os.path.join(base, "file.txt")), True
+        )
+        self.assertEqual(
+            netius.common.is_sub_path(base, os.path.join(base, "sub", "file.txt")),
+            True,
+        )
+        self.assertEqual(netius.common.is_sub_path(base, os.path.dirname(base)), False)
+        self.assertEqual(netius.common.is_sub_path(base, base + "-backup"), False)
+        self.assertEqual(
+            netius.common.is_sub_path(base, os.path.join(base + "-backup", "file.txt")),
+            False,
+        )
+        self.assertEqual(
+            netius.common.is_sub_path(base, os.path.join(os.sep, "etc", "passwd")),
+            False,
+        )
+
+        # the root of the file system contains every path there is
+        root = os.path.abspath(os.sep)
+        self.assertEqual(
+            netius.common.is_sub_path(root, os.path.join(root, "etc", "passwd")),
+            True,
+        )
+
+    def test_is_sub_path_missing(self):
+        base = tempfile.mkdtemp()
+
+        try:
+            # a path that is not there yet (the one of a file that is about
+            # to be written) is judged by its name alone, as there is no
+            # link in it to be resolved
+            self.assertEqual(
+                netius.common.is_sub_path(base, os.path.join(base, "new.txt")), True
+            )
+            self.assertEqual(
+                netius.common.is_sub_path(
+                    base, os.path.join(base, "missing", "new.txt")
+                ),
+                True,
+            )
+        finally:
+            shutil.rmtree(base)
+
+    def test_is_sub_path_link(self):
+        base = tempfile.mkdtemp()
+        outside = tempfile.mkdtemp()
+
+        try:
+            self._store(os.path.join(outside, "secret.txt"))
+            self._store(os.path.join(base, "file.txt"))
+            os.makedirs(os.path.join(base, "sub"))
+            self._link(
+                os.path.join(outside, "secret.txt"), os.path.join(base, "link.txt")
+            )
+            self._link(os.path.join(base, "file.txt"), os.path.join(base, "inside.txt"))
+            self._link(outside, os.path.join(base, "sub", "link"))
+            self._link(os.pardir, os.path.join(base, "sub", "up"))
+
+            # a link is judged by its target, so one that leads out of the
+            # base path is not contained while one that stays under it is
+            self.assertEqual(
+                netius.common.is_sub_path(base, os.path.join(base, "link.txt")),
+                False,
+            )
+            self.assertEqual(
+                netius.common.is_sub_path(base, os.path.join(base, "inside.txt")),
+                True,
+            )
+
+            # the link may sit anywhere below the base path, what follows
+            # it in the path being reached through its target
+            self.assertEqual(
+                netius.common.is_sub_path(base, os.path.join(base, "sub", "link")),
+                False,
+            )
+            self.assertEqual(
+                netius.common.is_sub_path(
+                    base, os.path.join(base, "sub", "link", "secret.txt")
+                ),
+                False,
+            )
+
+            # a link that climbs and yet lands under the base path leads
+            # nowhere it should not, so what is reached through it counts
+            self.assertEqual(
+                netius.common.is_sub_path(
+                    base, os.path.join(base, "sub", "up", "file.txt")
+                ),
+                True,
+            )
+            self.assertEqual(
+                netius.common.is_sub_path(
+                    base, os.path.join(base, "sub", "up", "link.txt")
+                ),
+                False,
+            )
+
+            # the operator may publish through a link on purpose, in which
+            # case the name under the base path is enough
+            self.assertEqual(
+                netius.common.is_sub_path(
+                    base, os.path.join(base, "link.txt"), follow_links=True
+                ),
+                True,
+            )
+        finally:
+            shutil.rmtree(base)
+            shutil.rmtree(outside)
+
+    def test_is_sub_path_base_link(self):
+        target = tempfile.mkdtemp()
+        outside = tempfile.mkdtemp()
+        base = target + "-link"
+
+        try:
+            self._link(target, base)
+            self._store(os.path.join(target, "file.txt"))
+            self._store(os.path.join(outside, "secret.txt"))
+            self._link(
+                os.path.join(outside, "secret.txt"), os.path.join(target, "link.txt")
+            )
+
+            # a base path that is itself reached through a link resolves the
+            # same way on both sides, so what sits under it is contained
+            # while a link out of it is still refused
+            self.assertEqual(netius.common.is_sub_path(base, base), True)
+            self.assertEqual(
+                netius.common.is_sub_path(base, os.path.join(base, "file.txt")), True
+            )
+            self.assertEqual(
+                netius.common.is_sub_path(base, os.path.join(base, "link.txt")),
+                False,
+            )
+        finally:
+            os.remove(base)
+            shutil.rmtree(target)
+            shutil.rmtree(outside)
+
     def test_size_round_unit(self):
         result = netius.common.size_round_unit(209715200, space=True)
         self.assertEqual(result, "200 MB")
@@ -308,3 +457,18 @@ def test_verify_many(self):
                 (1 == 1, 1 == 2), exception=netius.NetiusError
             ),
         )
+
+    def _link(self, source, target):
+        # the creation of a link is reserved to a privileged user under
+        # some systems, in which case the case has nothing to run
+        try:
+            os.symlink(source, target)
+        except (AttributeError, NotImplementedError, OSError):
+            self.skipTest("Skipping test: symbolic links unavailable")
+
+    def _store(self, path, contents=b"contents"):
+        file = open(path, "wb")
+        try:
+            file.write(contents)
+        finally:
+            file.close()