Confine the broker socket to the user's cache (SonarCloud #41–#44)

tig-cli/src/tig_cli/server.py

@@ -19,6 +19,7 @@
 import array
 import errno
 import os
+import re
 import secrets
 import select
 import selectors
@@ -68,13 +69,44 @@ def host_address() -> str:
 # reads is slowed down rather than buffered without end.
 HIGH_WATER = 1 << 20
 
+# What Docker and Podman accept as a container name. It cannot be empty, hold
+# a separator or start with a dot, so it is always one path component of its
+# own - never ``..``.
+CONTAINER_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*")
+
+
+def broker_address(container: str, home: str) -> str:
+    """Where the broker for ``container`` listens, kept inside ``home``'s cache.
+
+    Both arrive as command-line arguments, so neither may steer the socket -
+    or the directory made for it, or the stale socket removed in its place -
+    anywhere but its own directory under ``<home>/.cache/tig``.
+
+    Raises:
+        OSError: ``container`` is not a container name or ``home`` is not a
+            directory.
+    """
+    if not CONTAINER_NAME.fullmatch(container):
+        raise OSError(errno.EINVAL, f"not a container name: {container!r}")
+    home = os.path.abspath(home)
+    if not os.path.isdir(home):
+        raise OSError(errno.ENOTDIR, f"not a directory: {home!r}")
+    root = os.path.join(home, ".cache", "tig")
+    address = os.path.abspath(socket_path(home, container))
+    if (
+        not address.startswith(root + os.sep)
+        or os.path.dirname(os.path.dirname(address)) != root
+    ):
+        raise OSError(errno.EINVAL, f"broker socket outside {root!r}: {address!r}")
+    return address
+
 
 class Broker:
     """Serves one container's commands until it is idle or its agent goes."""
 
     def __init__(self, container: str, home: str):
         self.container = container
-        self.address = socket_path(home, container)
+        self.address = broker_address(container, home)
         self.token = secrets.token_hex(16)
         self.selector = selectors.SelectSelector()
         self.jobs: dict[str, Job] = {}

tig-cli/tests/test_broker.py

@@ -473,3 +473,49 @@ def _wait_for(path, timeout=30):
     while not path.exists():
         assert time.monotonic() < deadline, f"{path} never appeared"
         time.sleep(0.02)
+
+
+@pytest.mark.parametrize(
+    "container",
+    ["", ".", "..", "../escaped", "a/../../escaped", "/tmp/escaped", ".hidden",
+     "-flag", "tig vicar", "tig\x00vicar"],
+)
+def test_a_broker_refuses_a_container_name_that_is_not_one(home, container):
+    """The socket and its directory never land outside the user's cache."""
+    with pytest.raises(OSError):
+        Broker(container, str(home))
+
+    assert list(home.iterdir()) == []
+
+
+def test_a_broker_refuses_a_home_that_is_not_a_directory(home):
+    missing = home / "missing"
+    not_a_directory = home / "file"
+    not_a_directory.write_text("")
+
+    for candidate in (missing, not_a_directory):
+        with pytest.raises(OSError):
+            Broker(CONTAINER, str(candidate))
+
+    assert sorted(path.name for path in home.iterdir()) == ["file"]
+
+
+def test_a_broker_s_address_is_its_own_directory_under_the_cache(home):
+    name = "tig-vicar-0123456789ab"
+    address = server.broker_address(name, str(home / "sub" / ".."))
+
+    assert address == broker.socket_path(str(home), name)
+    assert os.path.dirname(address) == str(home / ".cache" / "tig" / name)
+
+
+def test_a_relative_home_is_made_absolute(home, monkeypatch):
+    monkeypatch.chdir(home)
+
+    address = server.broker_address(CONTAINER, ".")
+
+    assert address == broker.socket_path(os.getcwd(), CONTAINER)
+
+
+def test_the_broker_command_exits_unavailable_on_a_traversing_container(home):
+    assert broker.main(["../../escaped", str(home)]) == 69
+    assert list(home.iterdir()) == []