Provide a private Unix listener for owner-metered Messages

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

tests/test_messages_gate.py

@@ -187,3 +187,37 @@ def test_fixed_origin_no_redirect_proxy_or_request_url(policy, url):
         MessagesServer(
             policy, DeniedMeter(), provider_key="dummy", upstream_url=url, allow_test_http=True
         )
+
+
+def test_unix_listener_private_directory_and_no_tcp(tmp_path, policy):
+    import http.client
+    import socket
+
+    from llm_connect.messages_gate import MessagesServer, RequestRefused
+
+    private = tmp_path / "private"
+    private.mkdir(mode=0o700)
+    path = private / "messages.sock"
+    server = MessagesServer(policy, DeniedMeter(), provider_key="dummy", unix_path=path)
+    assert path.stat().st_mode & 0o777 == 0o600
+    with pytest.raises(RequestRefused, match="no TCP"):
+        _ = server.port
+    with pytest.raises(OSError):
+        MessagesServer(policy, DeniedMeter(), provider_key="dummy", unix_path=path)
+    server.start()
+    try:
+        connection = http.client.HTTPConnection("localhost", timeout=3)
+        connection.sock = socket.socket(socket.AF_UNIX)
+        connection.sock.settimeout(3)
+        connection.sock.connect(str(path))
+        connection.request("POST", "/execute", "{}", {"Content-Type": "application/json"})
+        response = connection.getresponse()
+        assert response.status == 404
+        response.read()
+        connection.close()
+    finally:
+        server.stop()
+    assert not path.exists()
+    private.chmod(0o755)
+    with pytest.raises(RequestRefused, match="private"):
+        MessagesServer(policy, DeniedMeter(), provider_key="dummy", unix_path=path)