diff --git a/src/dotenv/main.py b/src/dotenv/main.py index 3123690a..83354428 100644 --- a/src/dotenv/main.py +++ b/src/dotenv/main.py @@ -1,3 +1,4 @@ +import errno import io import logging import os @@ -135,6 +136,22 @@ def get_key( return DotEnv(dotenv_path, verbose=True, encoding=encoding).get(key_to_get) +# `O_NOFOLLOW` is POSIX-only; on platforms without it the flag is a no-op and +# `rewrite` falls back to the previous behaviour. +_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) + +# Errors a platform may raise when `O_NOFOLLOW` refuses to open a symlink. +_SYMLINK_ERRNOS = frozenset( + e + for e in (getattr(errno, "ELOOP", None), getattr(errno, "EMLINK", None)) + if e is not None +) + + +def _opener_no_follow(file: str, flags: int) -> int: + return os.open(file, flags | _O_NOFOLLOW) + + @contextmanager def rewrite( path: StrPath, @@ -145,7 +162,14 @@ def rewrite( path = os.path.realpath(path) try: - source: IO[str] = open(path, encoding=encoding) + # Do not read through a symlink unless asked to: `os.replace` below + # replaces the link itself rather than its target, so the target's + # contents are not the contents of the file being written. + source: IO[str] = open( + path, + encoding=encoding, + opener=None if follow_symlinks else _opener_no_follow, + ) try: path_stat = os.lstat(path) original_mode: Optional[int] = ( @@ -159,6 +183,13 @@ def rewrite( except FileNotFoundError: source = io.StringIO("") original_mode = None + except OSError as exc: + if exc.errno not in _SYMLINK_ERRNOS: + raise + # The path is a symlink and we are not following it, so there is no + # existing content to carry over into its replacement. + source = io.StringIO("") + original_mode = None with tempfile.NamedTemporaryFile( mode="w", diff --git a/tests/test_main.py b/tests/test_main.py index 6f9d4c5c..ee65e18c 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -129,16 +129,19 @@ def tracking_open(*args, **kwargs): sys.platform == "win32", reason="symlinks require elevated privileges on Windows" ) def test_set_key_symlink_to_existing_file(tmp_path): + # The target holds a different key from the one being set, so that content + # read through the symlink would be visible in the result rather than + # overwritten by the assignment. target = tmp_path / "target.env" - target.write_text("a=x\n") + target.write_text("b=x\n") symlink = tmp_path / ".env" symlink.symlink_to(target) dotenv.set_key(symlink, "a", "y") - assert target.read_text() == "a=x\n" + assert target.read_text() == "b=x\n" assert not symlink.is_symlink() - assert "a='y'" in symlink.read_text() + assert symlink.read_text() == "a='y'\n" assert stat.S_IMODE(symlink.stat().st_mode) == 0o600 @@ -321,14 +324,16 @@ def test_unset_non_existent_file(tmp_path): sys.platform == "win32", reason="symlinks require elevated privileges on Windows" ) def test_unset_key_symlink_to_existing_file(tmp_path): + # As above, the target holds a different key from the one being unset, so + # that content read through the symlink would remain visible in the result. target = tmp_path / "target.env" - target.write_text("a=x\n") + target.write_text("b=x\n") symlink = tmp_path / ".env" symlink.symlink_to(target) dotenv.unset_key(symlink, "a") - assert target.read_text() == "a=x\n" + assert target.read_text() == "b=x\n" assert not symlink.is_symlink() assert symlink.read_text() == ""