Fix permission handling for an open file descriptor

- work in progress

pyfakefs/fake_file.py

@@ -53,6 +53,7 @@
     AnyPath,
     AnyString,
     get_locale_encoding,
+    _OpenModes,
 )
 
 if TYPE_CHECKING:
@@ -134,6 +135,7 @@ def __init__(
         encoding: Optional[str] = None,
         errors: Optional[str] = None,
         side_effect: Optional[Callable[["FakeFile"], None]] = None,
+        open_modes: Optional[_OpenModes] = None,
     ):
         """
         Args:
@@ -152,6 +154,7 @@ def __init__(
             errors: The error mode used for encoding/decoding errors.
             side_effect: function handle that is executed when file is written,
                 must accept the file object as an argument.
+            open_modes: The modes the file was opened with (e.g. can read, write etc.)
         """
         # to be backwards compatible regarding argument order, we raise on None
         if filesystem is None:
@@ -180,6 +183,7 @@ def __init__(
         # Linux specific: extended file system attributes
         self.xattr: Dict = {}
         self.opened_as: AnyString = ""
+        self.open_modes = open_modes
 
     @property
     def byte_contents(self) -> Optional[bytes]:
@@ -755,6 +759,7 @@ def __init__(
         errors: Optional[str],
         buffering: int,
         raw_io: bool,
+        opened_as_fd: bool,
         is_stream: bool = False,
     ):
         self.file_object = file_object
@@ -766,6 +771,7 @@ def __init__(
         self._file_epoch = file_object.epoch
         self.raw_io = raw_io
         self._binary = binary
+        self._opened_as_fd = opened_as_fd
         self.is_stream = is_stream
         self._changed = False
         self._buffer_size = buffering
@@ -844,8 +850,17 @@ def close(self) -> None:
             return
 
         # for raw io, all writes are flushed immediately
-        if self.allow_update and not self.raw_io:
-            self.flush()
+        if not self.raw_io:
+            try:
+                self.flush()
+            except OSError as e:
+                if e.errno == errno.EBADF:
+                    # if we get here, we have an open file descriptor
+                    # without write permission, which has to be closed
+                    assert self.filedes
+                    self._filesystem._close_open_file(self.filedes)
+                raise
+
             if self._filesystem.is_windows_fs and self._changed:
                 self.file_object.st_mtime = helpers.now()
 
@@ -880,8 +895,12 @@ def _try_flush(self, old_pos: int) -> None:
 
     def flush(self) -> None:
         """Flush file contents to 'disk'."""
+        if self.is_stream:
+            return
+
         self._check_open_file()
-        if self.allow_update and not self.is_stream:
+
+        if self.allow_update:
             contents = self._io.getvalue()
             if self._append:
                 self._sync_io()
@@ -901,9 +920,15 @@ def flush(self) -> None:
                     self.file_object.st_ctime = current_time
                     self.file_object.st_mtime = current_time
             self._file_epoch = self.file_object.epoch
-
-            if not self.is_stream:
-                self._flush_related_files()
+            self._flush_related_files()
+        else:
+            buf_length = len(self._io.getvalue())
+            content_length = 0
+            if self.file_object.byte_contents is not None:
+                content_length = len(self.file_object.byte_contents)
+            # an error is only raised if there is something to flush
+            if content_length != buf_length:
+                self._filesystem.raise_os_error(errno.EBADF)
 
     def update_flush_pos(self) -> None:
         self._flush_pos = self._io.tell()
@@ -1146,7 +1171,7 @@ def __getattr__(self, name: str) -> Any:
             self._check_open_file()
         if not self._read and reading:
             return self._read_error()
-        if not self.allow_update and writing:
+        if not self._opened_as_fd and not self.allow_update and writing:
             return self._write_error()
 
         if reading:

pyfakefs/fake_open.py

@@ -17,7 +17,6 @@
 import errno
 import os
 import sys
-from collections import namedtuple
 from stat import (
     S_ISDIR,
 )
@@ -43,17 +42,13 @@
     is_root,
     PERM_READ,
     PERM_WRITE,
+    _OpenModes,
 )
 
 if TYPE_CHECKING:
     from pyfakefs.fake_filesystem import FakeFilesystem
 
 
-_OpenModes = namedtuple(
-    "_OpenModes",
-    "must_exist can_read can_write truncate append must_not_exist",
-)
-
 _OPEN_MODE_MAP = {
     # mode name:(file must exist, can read, can write,
     #            truncate, append, must not exist)
@@ -148,6 +143,7 @@ def call(
             raise ValueError("binary mode doesn't take an encoding argument")
 
         newline, open_modes = self._handle_file_mode(mode, newline, open_modes)
+        opened_as_fd = isinstance(file_, int)
 
         # the pathlib opener is defined in a Path instance that may not be
         # patched under some circumstances; as it just calls standard open(),
@@ -157,7 +153,9 @@ def call(
             # here as if directly passed
             file_ = opener(file_, self._open_flags_from_open_modes(open_modes))
 
-        file_object, file_path, filedes, real_path = self._handle_file_arg(file_)
+        file_object, file_path, filedes, real_path, can_write = self._handle_file_arg(
+            file_
+        )
         if file_object is None and file_path is None:
             # file must be a fake pipe wrapper, find it...
             if (
@@ -197,7 +195,11 @@ def call(
 
         assert real_path is not None
         file_object = self._init_file_object(
-            file_object, file_path, open_modes, real_path
+            file_object,
+            file_path,
+            open_modes,
+            real_path,
+            check_file_permission=not opened_as_fd,
         )
 
         if S_ISDIR(file_object.st_mode):
@@ -218,7 +220,7 @@ def call(
         fakefile = FakeFileWrapper(
             file_object,
             file_path,
-            update=open_modes.can_write,
+            update=open_modes.can_write and can_write,
             read=open_modes.can_read,
             append=open_modes.append,
             delete_on_close=self._delete_on_close,
@@ -230,6 +232,7 @@ def call(
             errors=errors,
             buffering=buffering,
             raw_io=self.raw_io,
+            opened_as_fd=opened_as_fd,
         )
         if filedes is not None:
             fakefile.filedes = filedes
@@ -267,16 +270,25 @@ def _init_file_object(
         file_path: AnyStr,
         open_modes: _OpenModes,
         real_path: AnyString,
+        check_file_permission: bool,
     ) -> FakeFile:
         if file_object:
-            if not is_root() and (
-                (open_modes.can_read and not file_object.has_permission(PERM_READ))
-                or (open_modes.can_write and not file_object.has_permission(PERM_WRITE))
+            if (
+                check_file_permission
+                and not is_root()
+                and (
+                    (open_modes.can_read and not file_object.has_permission(PERM_READ))
+                    or (
+                        open_modes.can_write
+                        and not file_object.has_permission(PERM_WRITE)
+                    )
+                )
             ):
                 self.filesystem.raise_os_error(errno.EACCES, file_path)
             if open_modes.can_write:
                 if open_modes.truncate:
                     file_object.set_contents("")
+            file_object
         else:
             if open_modes.must_exist:
                 self.filesystem.raise_os_error(errno.ENOENT, file_path)
@@ -304,16 +316,21 @@ def _init_file_object(
 
     def _handle_file_arg(
         self, file_: Union[AnyStr, int]
-    ) -> Tuple[Optional[FakeFile], Optional[AnyStr], Optional[int], Optional[AnyStr]]:
+    ) -> Tuple[
+        Optional[FakeFile], Optional[AnyStr], Optional[int], Optional[AnyStr], bool
+    ]:
         file_object = None
         if isinstance(file_, int):
             # opening a file descriptor
             filedes: int = file_
             wrapper = self.filesystem.get_open_file(filedes)
+            can_write = True
             if isinstance(wrapper, FakePipeWrapper):
-                return None, None, filedes, None
+                return None, None, filedes, None, can_write
             if isinstance(wrapper, FakeFileWrapper):
                 self._delete_on_close = wrapper.delete_on_close
+                can_write = wrapper.allow_update
+
             file_object = cast(
                 FakeFile, self.filesystem.get_open_file(filedes).get_object()
             )
@@ -324,6 +341,7 @@ def _handle_file_arg(
                 cast(AnyStr, path),  # pytype: disable=invalid-annotation
                 filedes,
                 cast(AnyStr, path),  # pytype: disable=invalid-annotation
+                can_write,
             )
 
         # open a file file by path
@@ -337,7 +355,7 @@ def _handle_file_arg(
                 file_object = self.filesystem.get_object_from_normpath(
                     real_path, check_read_perm=False
                 )
-        return file_object, file_path, None, real_path
+        return file_object, file_path, None, real_path, True
 
     def _handle_file_mode(
         self,

pyfakefs/helpers.py

@@ -19,6 +19,7 @@
 import stat
 import sys
 import time
+from collections import namedtuple
 from copy import copy
 from stat import S_IFLNK
 from typing import Union, Optional, Any, AnyStr, overload, cast
@@ -37,6 +38,11 @@
 PERM_DEF_FILE = 0o666  # Default permission bits (regular file)
 PERM_ALL = 0o7777  # All permission bits.
 
+_OpenModes = namedtuple(
+    "_OpenModes",
+    "must_exist can_read can_write truncate append must_not_exist",
+)
+
 if sys.platform == "win32":
     USER_ID = 1
     GROUP_ID = 1

pyfakefs/tests/fake_open_test.py

@@ -959,6 +959,14 @@ def test_use_opener_with_read(self):
             with self.assertRaises(OSError):
                 f.write("foo")
 
+    def test_no_opener_with_read(self):
+        file_path = self.make_path("foo")
+        self.create_file(file_path, contents="test")
+        with self.open(file_path) as f:
+            assert f.read() == "test"
+            with self.assertRaises(OSError):
+                f.write("foo")
+
     def test_use_opener_with_read_plus(self):
         file_path = self.make_path("foo")
         self.create_file(file_path, contents="test")