fix: prevent path traversal (CWE-22) in corpus and data path functions

AGENTS.md

@@ -6,7 +6,7 @@
       <https://github.com/PyThaiNLP/pythainlp/blob/dev/tests/README.md>.
       The document list test categories, their dependency sets,
       and test naming conventions.
-- [ ] Use reStructuredText for docstring (PEP 287), targetting Sphinx.
+- [ ] Use reStructuredText for docstring (PEP 287), targeting Sphinx.
 - [ ] When possible, follow NLTK established convention of submodule
       name (tend to be a verb or a generic noun), function name, and
       configuration. Communicate this to the users during code review.
@@ -20,7 +20,7 @@
       the repo. Read its usage and information it generates at
       <https://github.com/PyThaiNLP/pythainlp/blob/dev/build_tools/analysis/README.md>.
       Mind that the analyzer can create false positives,
-      please refer to Python tyep specification when in doubt.
+      please refer to Python type specification when in doubt.
 - [ ] Complete type annotations for function, method, class, variable, etc.
       Maintain near-100% type annotation coverage.
 - [ ] Add tests for new functionality or behavior.
@@ -31,6 +31,9 @@
 - [ ] Major changes should be logged in the change log at
       <https://github.com/PyThaiNLP/pythainlp/blob/dev/CHANGELOG.md>.
       Provide issue number or PR number if available.
+- [ ] Do not use os.path.join();
+      always use pythainlp.tools.safe_path_join() instead,
+      to prevent path traversal vulnerabilities (CWE-22).
 
 ## Project contribution guidelines
 

CHANGELOG.md

@@ -15,7 +15,16 @@ and this project adheres to
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 - Full release notes: <https://github.com/PyThaiNLP/pythainlp/releases>
-- Commit history: <https://github.com/PyThaiNLP/pythainlp/compare/v5.3.0...v5.3.1>
+- Commit history: <https://github.com/PyThaiNLP/pythainlp/compare/v5.3.1...v5.3.2>
+
+## [5.3.2] - 2026-03-20
+
+This release focuses on security issues related to path traversal.
+
+### Security
+
+- Prevent path traversal: validate that paths stay within
+  their expected base directory (#1342)
 
 ## [5.3.1] - 2026-03-14
 
@@ -212,6 +221,7 @@ The minimum requirement is now Python 3.9.
 
 - See <https://github.com/PyThaiNLP/pythainlp/releases/tag/v5.0.0>
 
+[5.3.2]: https://github.com/PyThaiNLP/pythainlp/compare/v5.3.1...v5.3.2
 [5.3.1]: https://github.com/PyThaiNLP/pythainlp/compare/v5.3.0...v5.3.1
 [5.3.0]: https://github.com/PyThaiNLP/pythainlp/compare/v5.2.0...v5.3.0
 [5.2.0]: https://github.com/PyThaiNLP/pythainlp/compare/v5.1.2...v5.2.0

pythainlp/corpus/__init__.py

@@ -22,9 +22,7 @@
     "get_corpus_db_detail",
     "get_corpus_default_db",
     "get_corpus_path",
-    "get_path_folder_corpus",
     "get_hf_hub",
-    "path_pythainlp_corpus",
     "provinces",
     "remove",
     "thai_dict",
@@ -89,9 +87,7 @@ def corpus_db_path() -> str:
     get_corpus_default_db,
     get_corpus_path,
     get_hf_hub,
-    get_path_folder_corpus,
     make_safe_directory_name,
-    path_pythainlp_corpus,
     remove,
 )
 from pythainlp.corpus.common import (

pythainlp/corpus/core.py

@@ -18,7 +18,11 @@
 from pythainlp import __version__
 from pythainlp.corpus import corpus_db_path, corpus_db_url, corpus_path
 from pythainlp.tools import get_full_data_path
-from pythainlp.tools.path import is_offline_mode, is_read_only_mode
+from pythainlp.tools.path import (
+    is_offline_mode,
+    is_read_only_mode,
+    safe_path_join,
+)
 
 if TYPE_CHECKING:
     from http.client import HTTPMessage, HTTPResponse
@@ -103,16 +107,6 @@ def get_corpus_db_detail(name: str, version: str = "") -> dict[str, Any]:
     return {}
 
 
-def path_pythainlp_corpus(filename: str) -> str:
-    """Get path pythainlp.corpus data
-
-    :param str filename: filename of the corpus to be read
-
-    :return: : path of corpus
-    :rtype: str
-    """
-    return os.path.join(corpus_path(), filename)
-
 
 @lru_cache(maxsize=None)
 def get_corpus(filename: str, comments: bool = True) -> frozenset[str]:
@@ -245,13 +239,15 @@ def get_corpus_default_db(name: str, version: str = "") -> Optional[str]:
 
     if name in corpus_db:
         if version in corpus_db[name]["versions"]:
-            return path_pythainlp_corpus(
-                corpus_db[name]["versions"][version]["filename"]
+            return safe_path_join(
+                corpus_path(),
+                corpus_db[name]["versions"][version]["filename"],
             )
         elif not version:  # load latest version
             version = corpus_db[name]["latest_version"]
-            return path_pythainlp_corpus(
-                corpus_db[name]["versions"][version]["filename"]
+            return safe_path_join(
+                corpus_path(),
+                corpus_db[name]["versions"][version]["filename"],
             )
 
     return None
@@ -852,17 +848,6 @@ def remove(name: str) -> bool:
     return False
 
 
-def get_path_folder_corpus(name: str, version: str, *path: str) -> str:
-    corpus_path = get_corpus_path(name, version)
-    if not corpus_path:
-        raise FileNotFoundError(
-            f"corpus-not-found name={name!r} version={version!r}\n"
-            f"  Corpus '{name}' (version {version}) not found.\n"
-            f"    Python: pythainlp.corpus.download('{name}')\n"
-            f"    CLI:    thainlp data get {name}"
-        )
-    return os.path.join(corpus_path, *path)
-
 
 def make_safe_directory_name(name: str) -> str:
     """Make safe directory name

pythainlp/tag/wangchanberta_onnx.py

@@ -11,7 +11,8 @@
     import sentencepiece as spm
     from onnxruntime import InferenceSession, SessionOptions
 
-from pythainlp.corpus import get_path_folder_corpus
+from pythainlp.corpus import get_corpus_path
+from pythainlp.tools import safe_path_join
 
 
 class WngchanBerta_ONNX:
@@ -47,24 +48,21 @@ def __init__(
         self.options.graph_optimization_level = (
             GraphOptimizationLevel.ORT_ENABLE_ALL
         )
+        _corpus_base = get_corpus_path(self.model_name, self.model_version)
+        if not _corpus_base:
+            raise FileNotFoundError(self.model_name)
         self.session = InferenceSession(
-            get_path_folder_corpus(
-                self.model_name, self.model_version, file_onnx
-            ),
+            safe_path_join(_corpus_base, file_onnx),
             sess_options=self.options,
             providers=providers,
         )
         self.session.disable_fallback()
         self.outputs_name = self.session.get_outputs()[0].name
         self.sp = spm.SentencePieceProcessor(
-            model_file=get_path_folder_corpus(
-                self.model_name, self.model_version, "sentencepiece.bpe.model"
-            )
+            model_file=safe_path_join(_corpus_base, "sentencepiece.bpe.model")
         )
         with open(
-            get_path_folder_corpus(
-                self.model_name, self.model_version, "config.json"
-            ),
+            safe_path_join(_corpus_base, "config.json"),
             encoding="utf-8-sig",
         ) as fh:
             self._json = json.load(fh)

pythainlp/tools/__init__.py

@@ -9,6 +9,7 @@
     "is_offline_mode",
     "is_read_only_mode",
     "is_unsafe_pickle_allowed",
+    "safe_path_join",
     "safe_print",
     "warn_deprecation",
 ]
@@ -22,4 +23,5 @@
     is_offline_mode,
     is_read_only_mode,
     is_unsafe_pickle_allowed,
+    safe_path_join,
 )

pythainlp/tools/path.py

@@ -159,13 +159,44 @@ def is_offline_mode() -> bool:
     return val.strip().lower() not in ("", "0", "false", "no", "off")
 
 
+def safe_path_join(base: str, *parts: str) -> str:
+    """Join *base* with *parts*, verify containment, and return the normalized path.
+
+    This is the authoritative path-traversal guard used throughout the library
+    wherever a base directory and external path components are combined
+    (e.g., :func:`get_full_data_path` and the internal corpus path helpers
+    in :mod:`pythainlp.corpus.core`).
+
+    :param str base: base directory that the result must reside within.
+    :param parts: additional path components to append.
+    :type parts: str
+
+    :return: normalized absolute path of the joined result.
+    :rtype: str
+
+    :raises ValueError: if the resolved path escapes *base*.
+    """
+    abs_base = os.path.abspath(base)
+    abs_full = os.path.abspath(os.path.join(abs_base, *parts))
+    if abs_full != abs_base and not abs_full.startswith(abs_base + os.sep):
+        raise ValueError(
+            f"Path traversal attempt detected: resolved path {abs_full!r} "
+            f"is outside the base directory {abs_base!r}."
+        )
+    return abs_full
+
+
 def get_full_data_path(path: str) -> str:
-    """This function joins path of :mod:`pythainlp` data directory and the
-    given path, and returns the full path.
+    """Join the PyThaiNLP data directory path with *path* and return the result.
+
+    :param str path: relative path or filename to append to the data directory.
 
-    :return: full path given the name of dataset
+    :return: normalized absolute path within the PyThaiNLP data directory.
     :rtype: str
 
+    :raises ValueError: if *path* resolves to a location outside the
+        PyThaiNLP data directory (path traversal attempt).
+
     :Example:
     ::
 
@@ -174,7 +205,7 @@ def get_full_data_path(path: str) -> str:
         get_full_data_path("ttc_freq.txt")
         # output: '/root/pythainlp-data/ttc_freq.txt'
     """
-    return os.path.join(get_pythainlp_data_path(), path)
+    return safe_path_join(get_pythainlp_data_path(), path)
 
 
 def get_pythainlp_data_path() -> str:

tests/core/test_security.py

@@ -11,11 +11,14 @@
 import unittest
 import zipfile
 
+from pythainlp.corpus import corpus_path
 from pythainlp.corpus.core import (
     _is_within_directory,
     _safe_extract_tar,
     _safe_extract_zip,
 )
+from pythainlp.tools import get_full_data_path, safe_path_join
+from pythainlp.tools.path import get_pythainlp_data_path
 
 
 class SecurityTestCase(unittest.TestCase):
@@ -186,6 +189,47 @@ def test_is_within_directory_with_symlinks(self):
             # in the _safe_extract_tar and _safe_extract_zip functions,
             # which check where symlinks actually point to.
 
+    def test_get_full_data_path_safe(self):
+        """Test that get_full_data_path returns a path within the data directory."""
+        result = get_full_data_path("ttc_freq.txt")
+        self.assertTrue(_is_within_directory(get_pythainlp_data_path(), result))
+
+    def test_get_full_data_path_rejects_traversal(self):
+        """Test that get_full_data_path rejects path traversal attempts."""
+        with self.assertRaises(ValueError) as ctx:
+            get_full_data_path("../../etc/passwd")
+        self.assertIn("path traversal", str(ctx.exception).lower())
+
+    def test_get_full_data_path_rejects_multiple_traversal(self):
+        """Test that get_full_data_path rejects multiple parent directory traversal."""
+        with self.assertRaises(ValueError) as ctx:
+            get_full_data_path("../../../root/.ssh/id_rsa")
+        self.assertIn("path traversal", str(ctx.exception).lower())
+
+    def test_safe_path_join_bundled_corpus_safe(self):
+        """Test safe_path_join with corpus_path() base accepts safe filenames."""
+        result = safe_path_join(corpus_path(), "negations_th.txt")
+        self.assertTrue(_is_within_directory(corpus_path(), result))
+
+    def test_safe_path_join_bundled_corpus_rejects_traversal(self):
+        """Test safe_path_join with corpus_path() base rejects traversal."""
+        with self.assertRaises(ValueError) as ctx:
+            safe_path_join(corpus_path(), "../../etc/passwd")
+        self.assertIn("path traversal", str(ctx.exception).lower())
+
+    def test_safe_path_join_with_tmpdir_safe(self):
+        """Test safe_path_join accepts safe sub-paths within a temp directory."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            result = safe_path_join(tmpdir, "model.txt")
+            self.assertTrue(_is_within_directory(tmpdir, result))
+
+    def test_safe_path_join_with_tmpdir_rejects_traversal(self):
+        """Test safe_path_join rejects traversal escape from a temp directory."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            with self.assertRaises(ValueError) as ctx:
+                safe_path_join(tmpdir, "../../etc/passwd")
+            self.assertIn("path traversal", str(ctx.exception).lower())
+
 
 if __name__ == "__main__":
     unittest.main()