From 5d8812b69c8fe42c1a38956749334c951344ef13 Mon Sep 17 00:00:00 2001
From: gshmu <mhohai@aim.com>
Date: Wed, 8 Apr 2026 10:15:37 +0800
Subject: [PATCH] feat: warn if running as root

---
 invoke/program.py | 23 ++++++++++++++++++++++-
 tests/program.py  | 41 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 63 insertions(+), 1 deletion(-)

diff --git a/invoke/program.py b/invoke/program.py
index 67b879b6..5893cb85 100644
--- a/invoke/program.py
+++ b/invoke/program.py
@@ -21,7 +21,7 @@
 from .exceptions import CollectionNotFound, Exit, ParseError, UnexpectedExit
 from .parser import Argument, Parser, ParserContext
 from .terminals import pty_size
-from .util import debug, enable_logging, helpline
+from .util import debug, enable_logging, helpline, isatty
 
 if TYPE_CHECKING:
     from .loader import Loader
@@ -186,6 +186,10 @@ def task_args(self) -> List["Argument"]:
     indent_width = 4
     indent = " " * indent_width
     col_padding = 3
+    root_warning = (
+        "WARNING: Running Invoke as root may create root-owned files and "
+        "cause later I/O or permission errors. Re-run as a non-root user."
+    )
 
     def __init__(
         self,
@@ -373,6 +377,7 @@ def run(self, argv: Optional[List[str]] = None, exit: bool = True) -> None:
         .. versionadded:: 1.0
         """
         try:
+            self.warn_if_running_as_root(is_testing=not exit)
             # Create an initial config, which will hold defaults & values from
             # most config file locations (all but runtime.) Used to inform
             # loading & parsing behavior.
@@ -421,6 +426,22 @@ def run(self, argv: Optional[List[str]] = None, exit: bool = True) -> None:
         except KeyboardInterrupt:
             sys.exit(1)  # Same behavior as Python itself outside of REPL
 
+    def warn_if_running_as_root(self, is_testing: bool = False) -> None:
+        """
+        Emit a warning when Invoke is executed as the root user.
+        """
+        if is_testing or not isatty(sys.stderr) or not self.running_as_root():
+            return
+        print(self.root_warning, file=sys.stderr)
+
+    def running_as_root(self) -> bool:
+        """
+        Return ``True`` when the current process is running as root.
+        """
+        if hasattr(os, "geteuid"):
+            return os.geteuid() == 0
+        return getpass.getuser() == "root"
+
     def parse_core(self, argv: Optional[List[str]]) -> None:
         debug("argv given to Program.run: {!r}".format(argv))
         self.normalize_argv(argv)
diff --git a/tests/program.py b/tests/program.py
index 3b6765cd..6aed4628 100644
--- a/tests/program.py
+++ b/tests/program.py
@@ -105,6 +105,47 @@ def write_pyc_explicitly_enables_bytecode_writing(self):
             expect("--write-pyc -c foo mytask")
             assert not sys.dont_write_bytecode
 
+        class root_warning:
+            @trap
+            def prints_warning_to_tty_stderr(self):
+                program = Program()
+                with patch.object(program, "running_as_root", return_value=True):
+                    sys.stderr.isatty = Mock(return_value=True)
+                    program.warn_if_running_as_root(is_testing=False)
+                assert (
+                    sys.stderr.getvalue()
+                    == "WARNING: Running Invoke as root may create root-owned files and cause later I/O or permission errors. Re-run as a non-root user.\n"
+                )
+
+            @trap
+            def does_not_warn_when_stderr_is_not_a_tty(self):
+                program = Program()
+                with patch.object(program, "running_as_root", return_value=True):
+                    sys.stderr.isatty = Mock(return_value=False)
+                    program.warn_if_running_as_root(is_testing=False)
+                assert sys.stderr.getvalue() == ""
+
+            @trap
+            def skips_warning_for_exit_false(self):
+                program = Program()
+                with patch.object(program, "running_as_root", return_value=True):
+                    sys.stderr.isatty = Mock(return_value=True)
+                    program.warn_if_running_as_root(is_testing=True)
+                assert sys.stderr.getvalue() == ""
+
+            @patch("invoke.program.os")
+            def uses_geteuid_when_available(self, os_):
+                os_.geteuid.return_value = 0
+                assert Program().running_as_root() is True
+
+            @patch("invoke.program.os", spec=[])
+            @patch("invoke.program.getpass.getuser")
+            def falls_back_to_username_when_geteuid_is_missing(
+                self, getuser
+            ):
+                getuser.return_value = "root"
+                assert Program().running_as_root() is True
+
     class normalize_argv:
         @patch("invoke.program.sys")
         def defaults_to_sys_argv(self, mock_sys):