From 3fbba98d80cb3c6bfacf708923c79b9ee8a1489c Mon Sep 17 00:00:00 2001 From: Bin Cao Date: Wed, 29 Apr 2026 11:13:56 +0800 Subject: [PATCH] Fix symlink following in set_key/unset_key python-dotenv reads key-value pairs from a .env file and can set them as environment variables. set_key() and unset_key() follow symbolic links when rewriting .env files via shutil.move(), allowing a local attacker to overwrite arbitrary files via a crafted symlink when a cross-device rename fallback is triggered. Fix by replacing shutil.move() with os.replace() and creating the temp file in the same directory as the target to ensure atomic same-device rename. Also preserve the original file mode and avoid blindly following symlinks. Add follow_symlinks parameter to rewrite(), set_key(), and unset_key() to allow opting in to the old behavior when needed. Backported from upstream commit 790c5c02991100aa1bf41ee5330aca75edc51311 to v1.1.0. CVE: CVE-2026-28684 Upstream-Status: Backport [https://github.com/theskumar/python-dotenv/commit/790c5c02991100aa1bf41ee5330aca75edc51311] Signed-off-by: Bin Cao --- src/dotenv/cli.py | 15 +++++- src/dotenv/main.py | 72 ++++++++++++++++++++----- tests/test_main.py | 129 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 201 insertions(+), 15 deletions(-) diff --git a/src/dotenv/cli.py b/src/dotenv/cli.py index 33ae148..259345b 100644 --- a/src/dotenv/cli.py +++ b/src/dotenv/cli.py @@ -93,7 +93,13 @@ def list(ctx: click.Context, format: bool) -> None: @click.argument('key', required=True) @click.argument('value', required=True) def set(ctx: click.Context, key: Any, value: Any) -> None: - """Store the given key/value.""" + """ + Store the given key/value. + + This doesn't follow symlinks, to avoid accidentally modifying a file at a + potentially untrusted path. + """ + file = ctx.obj['FILE'] quote = ctx.obj['QUOTE'] export = ctx.obj['EXPORT'] @@ -125,7 +131,12 @@ def get(ctx: click.Context, key: Any) -> None: @click.pass_context @click.argument('key', required=True) def unset(ctx: click.Context, key: Any) -> None: - """Removes the given key.""" + """ + Removes the given key. + + This doesn't follow symlinks, to avoid accidentally modifying a file at a + potentially untrusted path. + """ file = ctx.obj['FILE'] quote = ctx.obj['QUOTE'] success, key = unset_key(file, key, quote) diff --git a/src/dotenv/main.py b/src/dotenv/main.py index 1848d60..821bb9b 100644 --- a/src/dotenv/main.py +++ b/src/dotenv/main.py @@ -2,7 +2,7 @@ import io import logging import os import pathlib -import shutil +import stat import sys import tempfile from collections import OrderedDict @@ -13,9 +13,7 @@ from .parser import Binding, parse_stream from .variables import parse_variables # A type alias for a string path to be used for the paths in this file. -# These paths may flow to `open()` and `shutil.move()`; `shutil.move()` -# only accepts string paths, not byte paths or file descriptors. See -# https://github.com/python/typeshed/pull/6832. +# These paths may flow to `open()` and `os.replace()`. StrPath = Union[str, "os.PathLike[str]"] logger = logging.getLogger(__name__) @@ -131,21 +129,54 @@ def get_key( def rewrite( path: StrPath, encoding: Optional[str], + follow_symlinks: bool = False, ) -> Iterator[Tuple[IO[str], IO[str]]]: - pathlib.Path(path).touch() + if follow_symlinks: + path = os.path.realpath(path) - with tempfile.NamedTemporaryFile(mode="w", encoding=encoding, delete=False) as dest: + try: + source: IO[str] = open(path, encoding=encoding) + try: + path_stat = os.lstat(path) + original_mode: Optional[int] = ( + stat.S_IMODE(path_stat.st_mode) + if stat.S_ISREG(path_stat.st_mode) + else None + ) + except BaseException: + source.close() + raise + except FileNotFoundError: + source = io.StringIO("") + original_mode = None + + with tempfile.NamedTemporaryFile( + mode="w", + encoding=encoding, + delete=False, + prefix=".tmp_", + dir=os.path.dirname(os.path.abspath(path)), + ) as dest: + dest_path = pathlib.Path(dest.name) error = None + try: - with open(path, encoding=encoding) as source: + with source: yield (source, dest) except BaseException as err: error = err if error is None: - shutil.move(dest.name, path) + try: + if original_mode is not None: + os.chmod(dest_path, original_mode) + + os.replace(dest_path, path) + except BaseException: + dest_path.unlink(missing_ok=True) + raise else: - os.unlink(dest.name) + dest_path.unlink(missing_ok=True) raise error from None @@ -156,12 +187,16 @@ def set_key( quote_mode: str = "always", export: bool = False, encoding: Optional[str] = "utf-8", + follow_symlinks: bool = False, ) -> Tuple[Optional[bool], str, str]: """ Adds or Updates a key/value to the given .env - If the .env path given doesn't exist, fails instead of risking creating - an orphan .env somewhere in the filesystem + The target .env file is created if it doesn't exist. + + This function doesn't follow symlinks by default, to avoid accidentally + modifying a file at a potentially untrusted path. If you don't need this + protection and need symlinks to be followed, use `follow_symlinks`. """ if quote_mode not in ("always", "auto", "never"): raise ValueError(f"Unknown quote_mode: {quote_mode}") @@ -179,7 +214,10 @@ def set_key( else: line_out = f"{key_to_set}={value_out}\n" - with rewrite(dotenv_path, encoding=encoding) as (source, dest): + with rewrite(dotenv_path, encoding=encoding, follow_symlinks=follow_symlinks) as ( + source, + dest, + ): replaced = False missing_newline = False for mapping in with_warn_for_invalid_lines(parse_stream(source)): @@ -202,19 +240,27 @@ def unset_key( key_to_unset: str, quote_mode: str = "always", encoding: Optional[str] = "utf-8", + follow_symlinks: bool = False, ) -> Tuple[Optional[bool], str]: """ Removes a given key from the given `.env` file. If the .env path given doesn't exist, fails. If the given key doesn't exist in the .env, fails. + + This function doesn't follow symlinks by default, to avoid accidentally + modifying a file at a potentially untrusted path. If you don't need this + protection and need symlinks to be followed, use `follow_symlinks`. """ if not os.path.exists(dotenv_path): logger.warning("Can't delete from %s - it doesn't exist.", dotenv_path) return None, key_to_unset removed = False - with rewrite(dotenv_path, encoding=encoding) as (source, dest): + with rewrite(dotenv_path, encoding=encoding, follow_symlinks=follow_symlinks) as ( + source, + dest, + ): for mapping in with_warn_for_invalid_lines(parse_stream(source)): if mapping.key == key_to_unset: removed = True diff --git a/tests/test_main.py b/tests/test_main.py index 2d63eec..fbae934 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,6 +1,7 @@ import io import logging import os +import stat import sys import textwrap from unittest import mock @@ -61,6 +62,86 @@ def test_set_key_encoding(dotenv_path): assert dotenv_path.read_text(encoding=encoding) == "a='é'\n" +@pytest.mark.skipif( + sys.platform == "win32", reason="file mode bits behave differently on Windows" +) +def test_set_key_preserves_file_mode(dotenv_path): + dotenv_path.write_text("a=x\n") + dotenv_path.chmod(0o640) + mode_before = stat.S_IMODE(dotenv_path.stat().st_mode) + + dotenv.set_key(dotenv_path, "a", "y") + + mode_after = stat.S_IMODE(dotenv_path.stat().st_mode) + assert mode_before == mode_after + + +def test_rewrite_closes_file_handle_on_lstat_failure(tmp_path): + dotenv_path = tmp_path / ".env" + dotenv_path.write_text("a=x\n") + real_open = open + opened_handles = [] + + def tracking_open(*args, **kwargs): + handle = real_open(*args, **kwargs) + opened_handles.append(handle) + return handle + + with mock.patch("dotenv.main.os.lstat", side_effect=FileNotFoundError): + with mock.patch("dotenv.main.open", side_effect=tracking_open): + dotenv.set_key(dotenv_path, "a", "x") + + assert opened_handles, "expected at least one file to be opened" + assert all(handle.closed for handle in opened_handles) + + +@pytest.mark.skipif( + sys.platform == "win32", reason="symlinks require elevated privileges on Windows" +) +def test_set_key_symlink_to_existing_file(tmp_path): + target = tmp_path / "target.env" + target.write_text("a=x\n") + symlink = tmp_path / ".env" + symlink.symlink_to(target) + + dotenv.set_key(symlink, "a", "y") + + assert target.read_text() == "a=x\n" + assert not symlink.is_symlink() + assert "a='y'" in symlink.read_text() + assert stat.S_IMODE(symlink.stat().st_mode) == 0o600 + + +@pytest.mark.skipif( + sys.platform == "win32", reason="symlinks require elevated privileges on Windows" +) +def test_set_key_symlink_to_missing_file(tmp_path): + target = tmp_path / "nx" + symlink = tmp_path / ".env" + symlink.symlink_to(target) + + dotenv.set_key(symlink, "a", "x") + + assert not target.exists() + assert not symlink.is_symlink() + assert symlink.read_text() == "a='x'\n" + + +@pytest.mark.skipif( + sys.platform == "win32", reason="symlinks require elevated privileges on Windows" +) +def test_set_key_follow_symlinks(tmp_path): + target = tmp_path / "target.env" + target.write_text("a=x\n") + symlink = tmp_path / ".env" + symlink.symlink_to(target) + + dotenv.set_key(symlink, "a", "y", follow_symlinks=True) + + assert target.read_text() == "a='y'\n" + assert symlink.is_symlink() + + def test_set_key_permission_error(dotenv_path): dotenv_path.chmod(0o000) @@ -188,6 +269,54 @@ def test_unset_non_existent_file(tmp_path): ) +@pytest.mark.skipif( + sys.platform == "win32", reason="symlinks require elevated privileges on Windows" +) +def test_unset_key_symlink_to_existing_file(tmp_path): + target = tmp_path / "target.env" + target.write_text("a=x\n") + symlink = tmp_path / ".env" + symlink.symlink_to(target) + + dotenv.unset_key(symlink, "a") + + assert target.read_text() == "a=x\n" + assert not symlink.is_symlink() + assert symlink.read_text() == "" + + +@pytest.mark.skipif( + sys.platform == "win32", reason="symlinks require elevated privileges on Windows" +) +def test_unset_key_symlink_to_missing_file(tmp_path): + target = tmp_path / "nx" + symlink = tmp_path / ".env" + symlink.symlink_to(target) + logger = logging.getLogger("dotenv.main") + + with mock.patch.object(logger, "warning") as mock_warning: + result = dotenv.unset_key(symlink, "a") + + assert result == (None, "a") + assert symlink.is_symlink() + mock_warning.assert_called_once() + + +@pytest.mark.skipif( + sys.platform == "win32", reason="symlinks require elevated privileges on Windows" +) +def test_unset_key_follow_symlinks(tmp_path): + target = tmp_path / "target.env" + target.write_text("a=b\n") + symlink = tmp_path / ".env" + symlink.symlink_to(target) + + dotenv.unset_key(symlink, "a", follow_symlinks=True) + + assert target.read_text() == "" + assert symlink.is_symlink() + + def prepare_file_hierarchy(path): """ Create a temporary folder structure like the following: -- 2.43.0