feat: implement native Unix Domain Socket support

deps/amqp_client/src/amqp_network_connection.erl

@@ -159,6 +159,8 @@ inet_address_preference() ->
         {ok, false} -> [inet, inet6]
     end.
 
+gethostaddr({local, Path}) ->
+    {{local, Path}, local};
 gethostaddr(Host) ->
     resolve_address(Host, inet_address_preference()).
 

deps/rabbit/priv/schema/rabbit.schema

@@ -23,7 +23,7 @@
 ]}.
 
 {mapping, "listeners.tcp.$name", "rabbit.tcp_listeners",[
-    {datatype, [integer, ip]}
+    {datatype, [integer, ip, string]}
 ]}.
 
 {translation, "rabbit.tcp_listeners",
@@ -46,7 +46,7 @@ end}.
 ]}.
 
 {mapping, "listeners.ssl.$name", "rabbit.ssl_listeners",[
-    {datatype, [integer, ip]}
+    {datatype, [integer, ip, string]}
 ]}.
 
 {translation, "rabbit.ssl_listeners",

deps/rabbit/src/rabbit_networking.erl

@@ -82,8 +82,9 @@
 -type family() :: atom().
 -type listener_config() :: ip_port() |
                            {hostname(), ip_port()} |
-                           {hostname(), ip_port(), family()}.
--type address() :: {inet:ip_address(), ip_port(), family()}.
+                           {hostname(), ip_port(), family()} |
+                           string().
+-type address() :: {inet:ip_address() | {local, string()}, ip_port(), family()}.
 -type name_prefix() :: atom().
 -type protocol() :: atom().
 -type label() :: string().
@@ -146,6 +147,8 @@ fix_ssl_options(Config) ->
 
 -spec tcp_listener_addresses(listener_config()) -> [address()].
 
+tcp_listener_addresses(Path) when is_list(Path) andalso length(Path) > 0 andalso (hd(Path) =:= $/ orelse hd(Path) =:= $.) ->
+    [{{local, Path}, 0, local}];
 tcp_listener_addresses(Port) when is_integer(Port) ->
     tcp_listener_addresses_auto(Port);
 tcp_listener_addresses({"auto", Port}) ->
@@ -650,6 +653,8 @@ tune_buffer_size_static(Sock) ->
 
 %%--------------------------------------------------------------------
 
+tcp_host({local, Path}) ->
+    rabbit_data_coercion:to_binary(Path);
 tcp_host(IPAddress) ->
     rabbit_net:tcp_host(IPAddress).
 

deps/rabbit/test/unix_domain_socket_SUITE.erl

@@ -0,0 +1,110 @@
+-module(unix_domain_socket_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+
+all() ->
+    [test_unix_domain_socket,
+     test_uds_publish_consume,
+     test_uds_connection_failure].
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config, rabbit_ct_broker_helpers:setup_steps() ++ rabbit_ct_client_helpers:setup_steps()).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config, rabbit_ct_client_helpers:teardown_steps() ++ rabbit_ct_broker_helpers:teardown_steps()).
+
+init_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_started(Config, Testcase),
+    Config.
+
+end_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_finished(Config, Testcase),
+    Config.
+
+%% -------------------------------------------------------------------
+%% Test Cases
+%% -------------------------------------------------------------------
+
+test_unix_domain_socket(Config) ->
+    %% Path for our temporary unix domain socket
+    SocketPath = "/tmp/rabbitmq-test-uds.sock",
+
+    %% Ensure it doesn't already exist from a crashed run
+    file:delete(SocketPath),
+
+    %% Dynamically start a new listener on the local unix socket
+    ok = rabbit_ct_broker_helpers:rpc(Config, rabbit_networking, start_tcp_listener, [SocketPath, 10]),
+
+    %% Let the listener start up
+    timer:sleep(500),
+
+    %% Connect using the Erlang AMQP Client over the Unix socket
+    rabbit_ct_broker_helpers:rpc(Config, application, set_env, [rabbit, loopback_users, []]),
+    ConnParams = #amqp_params_network{host = {local, SocketPath}, port = 0},
+    {ok, Conn} = amqp_connection:start(ConnParams),
+    {ok, Channel} = amqp_connection:open_channel(Conn),
+
+    %% Perform a basic AMQP operation (declare a queue) to verify the socket works
+    Queue = <<"test_uds_queue">>,
+    Declare = #'queue.declare'{queue = Queue, durable = true},
+    #'queue.declare_ok'{} = amqp_channel:call(Channel, Declare),
+
+    %% Clean up
+    ok = amqp_channel:close(Channel),
+    ok = amqp_connection:close(Conn),
+
+    %% Stop the listener
+    ok = rabbit_ct_broker_helpers:rpc(Config, rabbit_networking, stop_tcp_listener, [SocketPath]),
+    file:delete(SocketPath),
+    ok.
+
+test_uds_publish_consume(Config) ->
+    SocketPath = "/tmp/rabbitmq-test-uds-pubsub.sock",
+    file:delete(SocketPath),
+    ok = rabbit_ct_broker_helpers:rpc(Config, rabbit_networking, start_tcp_listener, [SocketPath, 10]),
+    timer:sleep(500),
+
+    rabbit_ct_broker_helpers:rpc(Config, application, set_env, [rabbit, loopback_users, []]),
+    ConnParams = #amqp_params_network{host = {local, SocketPath}, port = 0},
+    {ok, Conn} = amqp_connection:start(ConnParams),
+    {ok, Channel} = amqp_connection:open_channel(Conn),
+
+    Queue = <<"test_uds_pubsub_queue">>,
+    #'queue.declare_ok'{} = amqp_channel:call(Channel, #'queue.declare'{queue = Queue, durable = true}),
+
+    %% Publish a message
+    Payload = <<"Hello over Unix Sockets">>,
+    Publish = #'basic.publish'{exchange = <<>>, routing_key = Queue},
+    amqp_channel:cast(Channel, Publish, #amqp_msg{payload = Payload}),
+
+    %% Consume the message
+    #'basic.consume_ok'{consumer_tag = CTag} = amqp_channel:call(Channel, #'basic.consume'{queue = Queue, no_ack = true}),
+    receive
+        {#'basic.deliver'{consumer_tag = CTag}, #amqp_msg{payload = MsgPayload}} ->
+            Payload = MsgPayload
+    after 5000 ->
+        ct:fail(message_not_received)
+    end,
+
+    ok = amqp_channel:close(Channel),
+    ok = amqp_connection:close(Conn),
+    ok = rabbit_ct_broker_helpers:rpc(Config, rabbit_networking, stop_tcp_listener, [SocketPath]),
+    file:delete(SocketPath),
+    ok.
+
+test_uds_connection_failure(_Config) ->
+    SocketPath = "/tmp/rabbitmq-nonexistent-socket.sock",
+    file:delete(SocketPath),
+
+    ConnParams = #amqp_params_network{host = {local, SocketPath}, port = 0},
+    case amqp_connection:start(ConnParams) of
+        {error, _} -> ok;
+        Other -> ct:fail({expected_error, Other})
+    end,
+    ok.
+

deps/rabbit_common/src/rabbit_core_metrics.erl

@@ -424,7 +424,7 @@ update_auth_attempt(RemoteAddress, Username, Protocol, Incr) ->
         {ok, true} ->
             Addr = case inet:is_ip_address(RemoteAddress) of
                        true ->
-                           list_to_binary(inet:ntoa(RemoteAddress));
+                           list_to_binary(rabbit_misc:ntoa(RemoteAddress));
                        false ->
                            rabbit_data_coercion:to_binary(RemoteAddress)
                    end,

deps/rabbit_common/src/rabbit_misc.erl

@@ -490,7 +490,7 @@ ensure_ok({error, Reason}, ErrorTag) -> throw({error, {ErrorTag, Reason}}).
 tcp_name(Prefix, IPAddress, Port)
   when is_atom(Prefix) andalso is_number(Port) ->
     list_to_atom(
-      format("~w_~ts:~w", [Prefix, inet_parse:ntoa(IPAddress), Port])).
+      format("~w_~ts:~w", [Prefix, ntoa(IPAddress), Port])).
 
 format_inet_error(E) -> format("~w (~ts)", [E, format_inet_error0(E)]).
 
@@ -799,11 +799,15 @@ const(X) -> fun () -> X end.
 
 %% Format IPv4-mapped IPv6 addresses as IPv4, since they're what we see
 %% when IPv6 is enabled but not used (i.e. 99% of the time).
+ntoa({local, Path}) ->
+    Path;
 ntoa({0,0,0,0,0,16#ffff,AB,CD}) ->
     inet_parse:ntoa({AB bsr 8, AB rem 256, CD bsr 8, CD rem 256});
 ntoa(IP) ->
     inet_parse:ntoa(IP).
 
+ntoab({local, Path}) ->
+    Path;
 ntoab(IP) ->
     Str = ntoa(IP),
     case string:str(Str, ":") of

deps/rabbit_common/src/rabbit_net.erl

@@ -294,6 +294,7 @@ is_loopback(Sock) when is_port(Sock) ; ?IS_SSL(Sock) ->
 is_loopback({127,_,_,_})             -> true;
 is_loopback({0,0,0,0,0,0,0,1})       -> true;
 is_loopback({0,0,0,0,0,65535,AB,CD}) -> is_loopback(ipv4(AB, CD));
+is_loopback({local, _})              -> true;
 is_loopback(_)                       -> false.
 
 ipv4(AB, CD) -> {AB bsr 8, AB band 255, CD bsr 8, CD band 255}.

deps/rabbitmq_auth_backend_http/src/rabbit_auth_backend_http.erl

@@ -294,7 +294,7 @@ join_tags(Tags) ->
 parse_peeraddr(unknown) ->
     rabbit_data_coercion:to_list(unknown);
 parse_peeraddr(PeerAddr) ->
-    handle_inet_ntoa_peeraddr(inet:ntoa(PeerAddr), PeerAddr).
+    handle_inet_ntoa_peeraddr(rabbit_misc:ntoa(PeerAddr), PeerAddr).
 
 -spec handle_inet_ntoa_peeraddr({'error', term()} | string(), inet:ip_address() | unknown) -> string().
 handle_inet_ntoa_peeraddr({error, einval}, PeerAddr) ->

deps/rabbitmq_cli/lib/rabbitmq/cli/core/listeners.ex

@@ -74,7 +74,7 @@ defmodule RabbitMQ.CLI.Core.Listeners do
     %{
       node: node,
       protocol: protocol,
-      interface: :inet.ntoa(interface) |> to_string |> maybe_enquote_interface,
+      interface: format_interface(interface),
       port: port,
       purpose: protocol_label(to_atom(protocol))
     }
@@ -91,7 +91,7 @@ defmodule RabbitMQ.CLI.Core.Listeners do
     %{
       node: node,
       protocol: protocol,
-      interface: :inet.ntoa(interface) |> to_string |> maybe_enquote_interface,
+      interface: format_interface(interface),
       port: port,
       purpose: protocol_label(to_atom(protocol)),
       certfile: read_cert(Keyword.get(opts, :certfile)),
@@ -159,7 +159,7 @@ defmodule RabbitMQ.CLI.Core.Listeners do
     %{
       node: node,
       protocol: protocol,
-      interface: :inet.ntoa(interface) |> to_string |> maybe_enquote_interface,
+      interface: format_interface(interface),
       port: port,
       purpose: protocol_label(to_atom(protocol)),
       certfile: certfile |> to_string,
@@ -246,7 +246,7 @@ defmodule RabbitMQ.CLI.Core.Listeners do
       [
         node: node,
         protocol: protocol,
-        interface: :inet.ntoa(interface) |> to_string |> maybe_enquote_interface,
+        interface: format_interface(interface),
         port: port,
         purpose: protocol_label(to_atom(protocol))
       ]
@@ -350,6 +350,9 @@ defmodule RabbitMQ.CLI.Core.Listeners do
   # Implementation
   #
 
+  defp format_interface({:local, path}), do: to_string(path)
+  defp format_interface(interface), do: :inet.ntoa(interface) |> to_string |> maybe_enquote_interface
+
   defp maybe_enquote_interface(value) do
     # This simplistic way of distinguishing IPv6 addresses,
     # networks address ranges, etc actually works better

deps/rabbitmq_management/src/rabbit_mgmt_wm_health_check_certificate_expiration.erl

@@ -110,7 +110,7 @@ listener_expiring_within(#listener{node = Node, protocol = Protocol, ip_address
         _ ->
             #{node => Node,
               protocol => Protocol,
-              interface => list_to_binary(inet:ntoa(Interface)),
+              interface => list_to_binary(rabbit_misc:ntoa(Interface)),
               port => Port,
               certfile => list_to_binary(Certfile),
               cacertfile => list_to_binary(Cacertfile),

deps/rabbitmq_mqtt/src/rabbit_mqtt_util.erl

@@ -181,6 +181,6 @@ truncate_binary(Bin, Size)
   when is_binary(Bin) ->
     binary:part(Bin, 0, Size).
 
--spec ip_address_to_binary(inet:ip_address()) -> binary().
+-spec ip_address_to_binary(inet:ip_address() | {local, string()}) -> binary().
 ip_address_to_binary(IpAddress) ->
-    list_to_binary(inet:ntoa(IpAddress)).
+    list_to_binary(rabbit_misc:ntoa(IpAddress)).

deps/rabbitmq_web_dispatch/src/rabbit_cowboy_stream_h.erl

@@ -100,7 +100,7 @@ user_from_req(Req) ->
     end.
 
 fmt_ip(IP) when is_tuple(IP) ->
-    inet_parse:ntoa(IP).
+    rabbit_misc:ntoa(IP).
 
 format_time() ->
     {{Year, Month, Date}, {Hour, Min, Sec}} = calendar:local_time(),

deps/rabbitmq_web_stomp/src/rabbit_web_stomp_listener.erl

@@ -223,7 +223,7 @@ get_binding_address(Configuration) ->
         undefined ->
             "0.0.0.0";
         IP when is_tuple(IP) ->
-            inet:ntoa(IP);
+            rabbit_misc:ntoa(IP);
         IP when is_list(IP) ->
             IP
     end.