Merge pull request #1369 from PyThaiNLP/copilot/bug-fix-pythainlp-path-manipulation

fix: replace os.path.join with safe_path_join to prevent path manipulation (CWE-22)

Author: bact

CHANGELOG.md

@@ -38,6 +38,11 @@ and this project adheres to
 - thai2rom_onnx: fix ONNX encoder model and fix inference bugs (#1349)
 - wordnet: fix AttributeError (#1354)
 
+### Security
+
+- Replace `os.path.join` with `safe_path_join` throughout the codebase
+  to prevent path manipulation vulnerabilities (CWE-22) (#1369)
+
 ## [5.3.2] - 2026-03-19
 
 This release focuses on security improvements related to path traversal

pythainlp/corpus/__init__.py

@@ -44,14 +44,13 @@
     "make_safe_directory_name",
 ]
 
-import os
-
 from pythainlp.tools import get_full_data_path, get_pythainlp_path
+from pythainlp.tools.path import safe_path_join
 
 # Remote and local corpus databases
 
 _CORPUS_DIRNAME: str = "corpus"
-_CORPUS_PATH: str = os.path.join(get_pythainlp_path(), _CORPUS_DIRNAME)
+_CORPUS_PATH: str = safe_path_join(get_pythainlp_path(), _CORPUS_DIRNAME)
 _CORPUS_DB_URL: str = "https://pythainlp.org/pythainlp-corpus/db.json"
 
 # filename of local corpus catalog

pythainlp/corpus/core.py

@@ -489,8 +489,9 @@ def _safe_extract_tar(tar: tarfile.TarFile, path: str) -> None:
         # Manual validation for older Python versions
         for member in tar.getmembers():
             # Check the member's target path
-            member_path = os.path.join(path, member.name)
-            if not _is_within_directory(path, member_path):
+            try:
+                safe_path_join(path, member.name)
+            except ValueError:
                 raise ValueError(
                     f"Attempted path traversal in tar file: {member.name}"
                 )
@@ -500,21 +501,26 @@ def _safe_extract_tar(tar: tarfile.TarFile, path: str) -> None:
                 # Get the link target (can be absolute or relative)
                 link_target = member.linkname
 
-                # If it's a relative symlink, resolve it relative to the member's directory
+                # If it's a relative symlink, resolve it relative to the member's directory.
+                # Pass the archive-relative dirname and link target as separate parts to
+                # safe_path_join, which canonicalises and validates containment in one step.
                 if not os.path.isabs(link_target):
-                    member_dir = os.path.dirname(member_path)
-                    link_target = os.path.join(member_dir, link_target)
+                    try:
+                        safe_path_join(
+                            path, os.path.dirname(member.name), link_target
+                        )
+                    except ValueError:
+                        raise ValueError(
+                            f"Symlink {member.name} points outside extraction directory: {member.linkname}"
+                        )
                 else:
                     # Absolute symlinks are dangerous - make them relative to extraction path
-                    link_target = os.path.join(
-                        path, link_target.lstrip(os.sep)
-                    )
-
-                # Check if the resolved symlink target is within the directory
-                if not _is_within_directory(path, link_target):
-                    raise ValueError(
-                        f"Symlink {member.name} points outside extraction directory: {member.linkname}"
-                    )
+                    try:
+                        safe_path_join(path, link_target.lstrip(os.sep))
+                    except ValueError:
+                        raise ValueError(
+                            f"Symlink {member.name} points outside extraction directory: {member.linkname}"
+                        )
 
         tar.extractall(path=path)
 
@@ -533,8 +539,9 @@ def _safe_extract_zip(zip_file: zipfile.ZipFile, path: str) -> None:
     created by Unix-based archiving tools and may not be portable.
     """
     for member in zip_file.namelist():
-        member_path = os.path.join(path, member)
-        if not _is_within_directory(path, member_path):
+        try:
+            safe_path_join(path, member)
+        except ValueError:
             raise ValueError(f"Attempted path traversal in zip file: {member}")
 
         # Check for potential symlinks in ZIP files
@@ -548,21 +555,26 @@ def _safe_extract_zip(zip_file: zipfile.ZipFile, path: str) -> None:
             # Read the symlink target from the file content
             link_target = zip_file.read(member).decode("utf-8")
 
-            # Resolve the link target relative to the member's directory
+            # Resolve the link target relative to the member's directory.
+            # Pass the archive-relative dirname and link target as separate parts to
+            # safe_path_join, which canonicalises and validates containment in one step.
             if not os.path.isabs(link_target):
-                member_dir = os.path.dirname(member_path)
-                resolved_target = os.path.join(member_dir, link_target)
+                try:
+                    safe_path_join(
+                        path, os.path.dirname(member), link_target
+                    )
+                except ValueError:
+                    raise ValueError(
+                        f"Symlink {member} points outside extraction directory: {link_target}"
+                    )
             else:
                 # Absolute symlinks - make them relative to extraction path
-                resolved_target = os.path.join(
-                    path, link_target.lstrip(os.sep)
-                )
-
-            # Check if the symlink target is within the directory
-            if not _is_within_directory(path, resolved_target):
-                raise ValueError(
-                    f"Symlink {member} points outside extraction directory: {link_target}"
-                )
+                try:
+                    safe_path_join(path, link_target.lstrip(os.sep))
+                except ValueError:
+                    raise ValueError(
+                        f"Symlink {member} points outside extraction directory: {link_target}"
+                    )
 
     zip_file.extractall(path=path)
 
@@ -909,7 +921,7 @@ def get_hf_hub(repo_id: str, filename: str = "") -> str:
         raise RuntimeError(f"An unexpected error occurred: {e}") from e
     hf_root = get_full_data_path("hf_models")
     name_dir = make_safe_directory_name(repo_id)
-    root_project = os.path.join(hf_root, name_dir)
+    root_project = safe_path_join(hf_root, name_dir)
     if filename:
         output_path = hf_hub_download(
             repo_id=repo_id, filename=filename, local_dir=root_project

pythainlp/parse/transformers_ud.py

@@ -21,6 +21,8 @@
         TokenClassificationPipeline,
     )
 
+from pythainlp.tools.path import safe_path_join
+
 
 class Parse:
     def __init__(
@@ -44,8 +46,8 @@ def __init__(
         x = AutoModelForTokenClassification.from_pretrained
         if os.path.isdir(model):
             d, t = (
-                x(os.path.join(model, "deprel")),
-                x(os.path.join(model, "tagger")),
+                x(safe_path_join(model, "deprel")),
+                x(safe_path_join(model, "tagger")),
             )
         else:
             c = AutoConfig.from_pretrained(

pythainlp/spell/words_spelling_correction.py

@@ -3,11 +3,11 @@
 # SPDX-License-Identifier: Apache-2.0
 from __future__ import annotations
 
-import os
 from importlib import import_module
 from typing import TYPE_CHECKING, Union, cast
 
 from pythainlp.corpus import get_hf_hub
+from pythainlp.tools.path import safe_path_join
 
 if TYPE_CHECKING:
     import numpy as np
@@ -88,10 +88,10 @@ def _load_embeddings(self) -> tuple[list[str], NDArray[np.float32]]:
         import numpy as np
 
         input_matrix = np.load(
-            os.path.join(self.model_dir, "embeddings.npy"), allow_pickle=False
+            safe_path_join(self.model_dir, "embeddings.npy"), allow_pickle=False
         )
         words = []
-        vocab_path = os.path.join(self.model_dir, "vocabulary.txt")
+        vocab_path = safe_path_join(self.model_dir, "vocabulary.txt")
         with open(vocab_path, encoding="utf-8") as f:
             for line in f.readlines():
                 words.append(line.rstrip())

pythainlp/tag/perceptron.py

@@ -5,25 +5,25 @@
 
 from __future__ import annotations
 
-import os
 from typing import Optional
 
 from pythainlp.corpus import corpus_path, get_corpus_path
 from pythainlp.tag import PerceptronTagger, blackboard, orchid
+from pythainlp.tools.path import safe_path_join
 
 _BLACKBOARD_NAME: str = "blackboard_pt_tagger"
 
 _ORCHID_FILENAME: str = "pos_orchid_perceptron.json"
-_ORCHID_PATH: str = os.path.join(corpus_path(), _ORCHID_FILENAME)
+_ORCHID_PATH: str = safe_path_join(corpus_path(), _ORCHID_FILENAME)
 
 _PUD_FILENAME: str = "pos_ud_perceptron-v0.2.json"
-_PUD_PATH: str = os.path.join(corpus_path(), _PUD_FILENAME)
+_PUD_PATH: str = safe_path_join(corpus_path(), _PUD_FILENAME)
 
 _TDTB_FILENAME: str = "tdtb-pt_tagger.json"
-_TDTB_PATH: str = os.path.join(corpus_path(), _TDTB_FILENAME)
+_TDTB_PATH: str = safe_path_join(corpus_path(), _TDTB_FILENAME)
 
 _TUD_FILENAME: str = "pos_tud_perceptron.json"
-_TUD_PATH: str = os.path.join(corpus_path(), _TUD_FILENAME)
+_TUD_PATH: str = safe_path_join(corpus_path(), _TUD_FILENAME)
 
 _BLACKBOARD_TAGGER: Optional[PerceptronTagger] = None
 _ORCHID_TAGGER: Optional[PerceptronTagger] = None

pythainlp/tag/unigram.py

@@ -6,25 +6,25 @@
 from __future__ import annotations
 
 import json
-import os
 from typing import Optional
 
 from pythainlp.corpus import corpus_path, get_corpus_path
 from pythainlp.tag import blackboard, orchid
+from pythainlp.tools.path import safe_path_join
 
 _ORCHID_FILENAME: str = "pos_orchid_unigram.json"
-_ORCHID_PATH: str = os.path.join(corpus_path(), _ORCHID_FILENAME)
+_ORCHID_PATH: str = safe_path_join(corpus_path(), _ORCHID_FILENAME)
 
 _PUD_FILENAME: str = "pos_ud_unigram-v0.2.json"
-_PUD_PATH: str = os.path.join(corpus_path(), _PUD_FILENAME)
+_PUD_PATH: str = safe_path_join(corpus_path(), _PUD_FILENAME)
 
 _TDTB_FILENAME: str = "tdtb-unigram_tagger.json"
-_TDTB_PATH: str = os.path.join(corpus_path(), _TDTB_FILENAME)
+_TDTB_PATH: str = safe_path_join(corpus_path(), _TDTB_FILENAME)
 
 _BLACKBOARD_NAME: str = "blackboard_unigram_tagger"
 
 _TUD_FILENAME: str = "pos_tud_unigram.json"
-_TUD_PATH: str = os.path.join(corpus_path(), _TUD_FILENAME)
+_TUD_PATH: str = safe_path_join(corpus_path(), _TUD_FILENAME)
 
 _ORCHID_TAGGER: Optional[dict[str, str]] = None
 _PUD_TAGGER: Optional[dict[str, str]] = None

pythainlp/tokenize/crfcut.py

@@ -17,12 +17,11 @@
 
 from __future__ import annotations
 
-import os
-
 import pycrfsuite
 
 from pythainlp.corpus import corpus_path
 from pythainlp.tokenize import word_tokenize
+from pythainlp.tools.path import safe_path_join
 
 _ENDERS: set[str] = {
     # ending honorifics
@@ -176,7 +175,7 @@ def _extract_features(
 
 _CRFCUT_DATA_FILENAME: str = "sentenceseg_crfcut.model"
 _tagger: pycrfsuite.Tagger = pycrfsuite.Tagger()  # pyright: ignore[reportAttributeAccessIssue]  # pyrefly: ignore[missing-attribute]
-_tagger.open(os.path.join(corpus_path(), _CRFCUT_DATA_FILENAME))
+_tagger.open(safe_path_join(corpus_path(), _CRFCUT_DATA_FILENAME))
 
 
 def segment(text: str) -> list[str]:

pythainlp/translate/en_th.py

@@ -10,7 +10,6 @@
 
 from __future__ import annotations
 
-import os
 import warnings
 from typing import Optional
 
@@ -29,6 +28,7 @@
     ) from e
 
 from pythainlp.corpus import download, get_corpus_path
+from pythainlp.tools.path import safe_path_join
 
 _EN_TH_MODEL_NAME: str = "scb_1m_en-th_moses"
 # SCB_1M-MT_OPUS+TBASE_en-th_moses-spm_130000-16000_v1.0.tar.gz
@@ -45,7 +45,7 @@ def _get_translate_path(model: str, *path: str) -> str:
     corpus_path = get_corpus_path(model, version="1.0")
     if not corpus_path:
         return ""
-    return os.path.join(corpus_path, *path)
+    return safe_path_join(corpus_path, *path)
 
 
 def _download_install(name: str) -> None: