Special-case files in /proc/self/fds/ (#180)

Files created with this mechanism should not be resolved by
pathlib.Path.resolve. The results are not a valid path to the file.

resolves #176

CHANGELOG.rst

@@ -10,6 +10,9 @@ Unreleased
 
 .. vendor-insert-here
 
+- Fix handling of file descriptors created using the ``/proc/self/fd/``
+  mechanism (:issue:`176`)
+
 0.19.0
 ------
 

src/check_jsonschema/utils.py

@@ -3,6 +3,7 @@
 import linecache
 import os
 import pathlib
+import re
 import traceback
 import typing as t
 import urllib.parse
@@ -13,6 +14,7 @@
 
 WINDOWS = os.name == "nt"
 
+PROC_FD_PATH_PATTERN = re.compile(r"/proc/(self|\d+)/fd/\d+")
 
 # this is a short list of schemes which will be recognized as being
 # schemes at all; anything else will not even be reported as an
@@ -82,6 +84,11 @@ def filename2path(filename: str) -> pathlib.Path:
 
         p = pathlib.Path(filename)
 
+    # if passed a file descriptor object, do not try to resolve it
+    # the resolution behavior when using zsh `<()` redirection seems to result in
+    # an incorrect path being used
+    if PROC_FD_PATH_PATTERN.fullmatch(filename):
+        return p
     return p.resolve()
 
 

tests/acceptance/test_special_filetypes.py

@@ -0,0 +1,75 @@
+import os
+import platform
+import sys
+import threading
+
+import pytest
+
+
+@pytest.mark.skipif(
+    platform.system() != "Linux", reason="test requires /proc/self/ mechanism"
+)
+@pytest.mark.skipif(sys.version_info < (3, 8), reason="test uses os.memfd_create")
+def test_schema_and_instance_in_memfds(run_line_simple):
+    """
+    create memory file descriptors and write schema and instance data into those
+    ensure the result works when the paths to those fds are passed on the CLI
+    """
+    schemafd = os.memfd_create("test_memfd_schema")
+    instancefd = os.memfd_create("test_memfd_instance")
+    try:
+        os.write(schemafd, b'{"type": "integer"}')
+        os.write(instancefd, b"42")
+
+        schema_path = f"/proc/self/fd/{schemafd}"
+        instance_path = f"/proc/self/fd/{instancefd}"
+
+        run_line_simple(["--schemafile", schema_path, instance_path])
+    finally:
+        os.close(schemafd)
+        os.close(instancefd)
+
+
+@pytest.mark.skipif(os.name != "posix", reason="test requires mkfifo")
+@pytest.mark.parametrize("check_succeeds", (True, False))
+def test_schema_and_instance_in_fifos(tmp_path, run_line, check_succeeds):
+    """
+    create fifos and write schema and instance data into those
+    ensure the result works when the paths to those fds are passed on the CLI
+    """
+    schema_path = tmp_path / "schema"
+    instance_path = tmp_path / "instance"
+    os.mkfifo(schema_path)
+    os.mkfifo(instance_path)
+
+    # execute FIFO writes as blocking writes in background threads
+    # nonblocking writes fail file existence if there's no reader, so using a FIFO
+    # requires some level of concurrency
+    def fifo_write(path, data):
+        fd = os.open(path, os.O_WRONLY)
+        try:
+            os.write(fd, data)
+        finally:
+            os.close(fd)
+
+    schema_thread = threading.Thread(
+        target=fifo_write, args=[schema_path, b'{"type": "integer"}']
+    )
+    schema_thread.start()
+    instance_data = b"42" if check_succeeds else b'"foo"'
+    instance_thread = threading.Thread(
+        target=fifo_write, args=[instance_path, instance_data]
+    )
+    instance_thread.start()
+
+    try:
+        result = run_line(
+            ["check-jsonschema", "--schemafile", str(schema_path), str(instance_path)]
+        )
+        if check_succeeds:
+            assert result.exit_code == 0
+        else:
+            assert result.exit_code == 1
+    finally:
+        schema_thread.join(timeout=0.1)
+        instance_thread.join(timeout=0.1)

tests/unit/test_utils.py

@@ -0,0 +1,34 @@
+import os
+import platform
+import sys
+
+import pytest
+
+from check_jsonschema.utils import filename2path
+
+
+@pytest.mark.skipif(
+    not (platform.system() == "Linux"), reason="test requires /proc/self/ mechanism"
+)
+@pytest.mark.skipif(sys.version_info < (3, 8), reason="test uses os.memfd_create")
+@pytest.mark.parametrize("use_pid_in_path", (True, False))
+def test_filename2path_on_memfd(use_pid_in_path):
+    """
+    create a memory file descriptor with a path in /proc/self/fd/
+    and then attempt to resolve that to an absolute Path object
+    the end result should be untouched
+
+    pathlib behavior is, for example,
+
+    >>> pathlib.Path("/proc/self/fd/4").resolve()
+    PosixPath('/memfd:myfd (deleted)')
+    """
+    testfd = os.memfd_create("test_filename2path")
+    try:
+        pid = os.getpid() if use_pid_in_path else "self"
+        filename = f"/proc/{pid}/fd/{testfd}"
+        path = filename2path(filename)
+
+        assert str(path) == filename
+    finally:
+        os.close(testfd)