don't follow symlinks or trust another user's file in FS.open

lark/lark.py

@@ -382,6 +382,11 @@ def __init__(self, grammar: 'Union[Grammar, str, IO[str]]', **options) -> None:
                 except FileNotFoundError:
                     # The cache file doesn't exist; parse and compose the grammar as normal
                     pass
+                except PermissionError as e:
+                    # FS.open refused the cache (a symlink, owned by another user, or
+                    # writable by others). This is an intentional refusal, not a crash,
+                    # so log a single line and rebuild from the grammar as normal.
+                    logger.warning("Failed to load cache due to a permissions error: %s. Rebuilding from the grammar.", e)
                 except Exception: # We should probably narrow done which errors we catch here.
                     logger.exception("Failed to load Lark from cache: %r. We will try to carry on.", cache_fn)
 
@@ -485,8 +490,11 @@ def __init__(self, grammar: 'Union[Grammar, str, IO[str]]', **options) -> None:
                     f.write(cache_sha256.encode('utf8') + b'\n')
                     pickle.dump(used_files, f)
                     self.save(f, _LOAD_ALLOWED_OPTIONS)
+            except PermissionError as e:
+                # FS.open refused the cache path (a symlink, or owned by another user).
+                logger.warning("Failed to save cache due to a permissions error: %s", e)
             except IOError as e:
-                logger.exception("Failed to save Lark to cache: %r.", cache_fn, e)
+                logger.exception("Failed to save Lark to cache: %r. (%s)", cache_fn, e)
 
     if __doc__:
         __doc__ += "\n\n" + LarkOptions.OPTIONS_DOC

lark/utils.py

@@ -1,5 +1,6 @@
 import unicodedata
 import os
+import errno
 from itertools import product
 from collections import deque
 from typing import Callable, Iterator, List, Optional, Tuple, Type, TypeVar, Union, Dict, Any, Sequence, Iterable, AbstractSet
@@ -305,6 +306,58 @@ def combine_alternatives(lists):
 except ImportError:
     _has_atomicwrites = False
 
+
+def _open_private(name, mode, **kwargs):
+    """Like open(), but refuses to follow a symlink, refuses a file that belongs to
+    another user, refuses to read a file that other users can write, and keeps the
+    file readable only by its owner.
+
+    The parser cache is stored in a shared temporary directory under a name that is
+    derived from the grammar, so another user on the same machine can predict it and
+    get there first.
+
+    A refused file raises ``PermissionError``, which the cache loader turns into a
+    warning and a normal rebuild rather than a failure.
+    """
+    writing = "w" in mode
+    if writing:
+        # No O_TRUNC here: we truncate only after the ownership check passes, so a file
+        # we end up refusing (e.g. a shared cache owned by someone else) is left intact
+        # instead of being emptied first.
+        flags = os.O_WRONLY | os.O_CREAT
+    else:
+        flags = os.O_RDONLY
+    if "b" in mode:
+        flags |= getattr(os, "O_BINARY", 0)
+    try:
+        fd = os.open(name, flags | getattr(os, "O_NOFOLLOW", 0), 0o600)
+    except OSError as e:
+        # O_NOFOLLOW reports a symlink as ELOOP (EMLINK on some BSDs). Treat that as a
+        # refusal like the ownership/permission checks below, so the caller sees one
+        # kind of "won't use this cache" error rather than a raw traceback.
+        if e.errno in (errno.ELOOP, getattr(errno, "EMLINK", -1)):
+            raise PermissionError("Refusing to open %r: it is a symlink" % name) from e
+        raise
+    try:
+        st = os.fstat(fd)
+        if hasattr(os, "geteuid"):
+            if st.st_uid != os.geteuid():
+                raise PermissionError("Refusing to use %r: it belongs to another user" % name)
+            # On read, don't trust a file that group or others can write, since it
+            # could have been tampered with even though we own it. Only write bits let
+            # another user do that, so read bits are fine to leave alone.
+            if not writing and st.st_mode & 0o022:
+                raise PermissionError("Refusing to read %r: other users can write to it" % name)
+        if writing:
+            if hasattr(os, "fchmod"):
+                os.fchmod(fd, 0o600)
+            os.ftruncate(fd, 0)
+        return os.fdopen(fd, mode, **kwargs)
+    except Exception:
+        os.close(fd)
+        raise
+
+
 class FS:
     exists = staticmethod(os.path.exists)
 
@@ -313,7 +366,7 @@ def open(name, mode="r", **kwargs):
         if _has_atomicwrites and "w" in mode:
             return atomicwrites.atomic_write(name, mode=mode, overwrite=True, **kwargs)
         else:
-            return open(name, mode, **kwargs)
+            return _open_private(name, mode, **kwargs)
 
 
 class fzset(frozenset):

tests/__main__.py

@@ -7,7 +7,7 @@
 
 from .test_trees import TestTrees
 from .test_tools import TestStandalone
-from .test_cache import TestCache
+from .test_cache import TestCache, TestCacheFile, TestCacheFilePortable
 from .test_grammar import TestGrammar
 from .test_reconstructor import TestReconstructor
 from .test_tree_forest_transformer import TestTreeForestTransformer

tests/test_cache.py

@@ -1,11 +1,18 @@
 from __future__ import absolute_import
 
 import logging
+import os
+import shutil
+import stat
+import tempfile
 from unittest import TestCase, main, skipIf
+from unittest.mock import patch
 
 from lark import Lark, Tree, Transformer, UnexpectedInput
 from lark.exceptions import ConfigurationError
 from lark.lexer import Lexer, Token
+from lark.utils import FS, _open_private
+import lark.utils as lark_utils
 import lark.lark as lark_module
 from lark.reconstruct import Reconstructor
 from . import test_reconstructor
@@ -229,5 +236,106 @@ def test_reconstruct(self):
         self.assertEqual(test_reconstructor._remove_ws(code), test_reconstructor._remove_ws(new))
 
 
+@skipIf(os.name != 'posix', "requires posix file permissions and symlinks")
+class TestCacheFile(TestCase):
+    # The automatic cache name is derived from the grammar and lives in a shared
+    # temporary directory, so another user can predict the path and create it first.
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmpdir, True)
+        self.cache_fn = os.path.join(self.tmpdir, 'cache.tmp')
+        self.other_fn = os.path.join(self.tmpdir, 'other')
+        with open(self.other_fn, 'wb') as f:
+            f.write(b'original')
+
+    def test_load_does_not_follow_symlink(self):
+        os.symlink(self.other_fn, self.cache_fn)
+        with self.assertRaises(OSError):
+            FS.open(self.cache_fn, 'rb').close()
+
+    def test_save_does_not_follow_symlink(self):
+        os.symlink(self.other_fn, self.cache_fn)
+        try:
+            with FS.open(self.cache_fn, 'wb') as f:
+                f.write(b'overwritten')
+        except OSError:
+            pass
+        with open(self.other_fn, 'rb') as f:
+            self.assertEqual(f.read(), b'original')
+
+    def test_save_refuses_other_users_file_without_truncating(self):
+        # A refused file must be left intact, not emptied first. This is the plain-open
+        # write path (atomicwrites has its own mkstemp+rename), so exercise it directly.
+        # We can't chown to another user without privileges, so fake the mismatch.
+        with open(self.cache_fn, 'wb') as f:
+            f.write(b'colleague-data')
+        os.chmod(self.cache_fn, 0o600)
+        with patch('os.geteuid', return_value=os.geteuid() + 1):
+            with self.assertRaises(OSError):
+                _open_private(self.cache_fn, 'wb')
+        with open(self.cache_fn, 'rb') as f:
+            self.assertEqual(f.read(), b'colleague-data')
+
+    def test_lark_warns_on_refused_cache_without_traceback(self):
+        # A refused cache path must not break parser construction, and both the load
+        # and the save refusal should surface as one warning each, not an exception log.
+        os.symlink(self.other_fn, self.cache_fn)
+        with self.assertLogs(lark_module.logger, level='WARNING') as cm:
+            Lark('start: "a"', parser='lalr', cache=self.cache_fn)
+        self.assertEqual([r.levelname for r in cm.records], ['WARNING'] * len(cm.records))
+        self.assertTrue(any('Failed to load cache' in r.getMessage() for r in cm.records))
+        if not lark_utils._has_atomicwrites:
+            # With atomicwrites the save replaces the symlink atomically instead of refusing.
+            self.assertTrue(any('Failed to save cache' in r.getMessage() for r in cm.records))
+        with open(self.other_fn, 'rb') as f:
+            self.assertEqual(f.read(), b'original')
+
+    def test_save_keeps_cache_private(self):
+        with FS.open(self.cache_fn, 'wb') as f:
+            f.write(b'data')
+        self.assertEqual(stat.S_IMODE(os.stat(self.cache_fn).st_mode) & 0o077, 0)
+
+    def test_load_refuses_group_or_world_writable(self):
+        with open(self.cache_fn, 'wb') as f:
+            f.write(b'data')
+        os.chmod(self.cache_fn, 0o666)
+        with self.assertRaises(OSError):
+            FS.open(self.cache_fn, 'rb').close()
+
+    def test_load_allows_group_readable(self):
+        # Only write bits let another user tamper, so a file we own that is merely
+        # group/world readable is still fine to load.
+        with open(self.cache_fn, 'wb') as f:
+            f.write(b'data')
+        os.chmod(self.cache_fn, 0o644)
+        with FS.open(self.cache_fn, 'rb') as f:
+            self.assertEqual(f.read(), b'data')
+
+
+class TestCacheFilePortable(TestCase):
+    # Runs everywhere, including Windows, where the posix-only checks above are skipped:
+    # _open_private must still round-trip a normal read/write.
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmpdir, True)
+        self.cache_fn = os.path.join(self.tmpdir, 'cache.tmp')
+
+    def test_open_roundtrip(self):
+        with FS.open(self.cache_fn, 'wb') as f:
+            f.write(b'roundtrip')
+        with FS.open(self.cache_fn, 'rb') as f:
+            self.assertEqual(f.read(), b'roundtrip')
+
+    def test_open_truncates_existing_on_write(self):
+        with FS.open(self.cache_fn, 'wb') as f:
+            f.write(b'longer original content')
+        with FS.open(self.cache_fn, 'wb') as f:
+            f.write(b'short')
+        with FS.open(self.cache_fn, 'rb') as f:
+            self.assertEqual(f.read(), b'short')
+
+
 if __name__ == '__main__':
     main()