diff --git a/AGENTS.md b/AGENTS.md index ab126798e..6d225a119 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,7 +6,7 @@ . 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 . 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 . 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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 99a351715..bb9b8e119 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,7 +15,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). - Full release notes: -- Commit history: +- Commit history: + +## [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 +[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 diff --git a/pythainlp/corpus/__init__.py b/pythainlp/corpus/__init__.py index d5b48250e..70a60ef00 100644 --- a/pythainlp/corpus/__init__.py +++ b/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 ( diff --git a/pythainlp/corpus/core.py b/pythainlp/corpus/core.py index e26ddf640..c2964ea57 100644 --- a/pythainlp/corpus/core.py +++ b/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 diff --git a/pythainlp/tag/wangchanberta_onnx.py b/pythainlp/tag/wangchanberta_onnx.py index 4b76f2d49..cf62104c0 100644 --- a/pythainlp/tag/wangchanberta_onnx.py +++ b/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) diff --git a/pythainlp/tools/__init__.py b/pythainlp/tools/__init__.py index fcb8d662d..cc718a93d 100644 --- a/pythainlp/tools/__init__.py +++ b/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, ) diff --git a/pythainlp/tools/path.py b/pythainlp/tools/path.py index 2bd954c7d..b92afa876 100644 --- a/pythainlp/tools/path.py +++ b/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: diff --git a/tests/core/test_security.py b/tests/core/test_security.py index aed704858..07095b942 100644 --- a/tests/core/test_security.py +++ b/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()