pid1: make MaxConnectionsPerSource= also work for AF_UNIX sockets

The setting currently puts limits on connections per IP address and
AF_UNIX CID. Let's extend it to cover AF_UNIX too, where it puts a limit
on connections per UID.

This is particularly useful for the various Accept=yes Varlink services
we now have, as it means, the number of per-user instance services
cannot grow without bounds.

man/systemd.socket.xml

@@ -470,9 +470,10 @@
 
       <varlistentry>
         <term><varname>MaxConnectionsPerSource=</varname></term>
-        <listitem><para>The maximum number of connections for a service per source IP address.
-        This is very similar to the <varname>MaxConnections=</varname> directive
-        above. Disabled by default.</para>
+        <listitem><para>The maximum number of connections for a service per source IP address (in case of
+        IPv4/IPv6), per source CID (in case of <constant>AF_VSOCK</constant>), or source UID (in case of
+        <constant>AF_UNIX</constant>). This is very similar to the <varname>MaxConnections=</varname>
+        directive above. Disabled by default.</para>
 
         <xi:include href="version-info.xml" xpointer="v232"/>
         </listitem>

src/core/socket.c

@@ -53,6 +53,7 @@ struct SocketPeer {
         Socket *socket;
         union sockaddr_union peer;
         socklen_t peer_salen;
+        struct ucred peer_cred;
 };
 
 static const UnitActiveState state_translation_table[_SOCKET_STATE_MAX] = {
@@ -420,6 +421,8 @@ static void peer_address_hash_func(const SocketPeer *s, struct siphash *state) {
                 siphash24_compress_typesafe(s->peer.in6.sin6_addr, state);
         else if (s->peer.sa.sa_family == AF_VSOCK)
                 siphash24_compress_typesafe(s->peer.vm.svm_cid, state);
+        else if (s->peer.sa.sa_family == AF_UNIX)
+                siphash24_compress_typesafe(s->peer_cred.uid, state);
         else
                 assert_not_reached();
 }
@@ -438,6 +441,8 @@ static int peer_address_compare_func(const SocketPeer *x, const SocketPeer *y) {
                 return memcmp(&x->peer.in6.sin6_addr, &y->peer.in6.sin6_addr, sizeof(x->peer.in6.sin6_addr));
         case AF_VSOCK:
                 return CMP(x->peer.vm.svm_cid, y->peer.vm.svm_cid);
+        case AF_UNIX:
+                return CMP(x->peer_cred.uid, y->peer_cred.uid);
         }
         assert_not_reached();
 }
@@ -466,16 +471,22 @@ static int socket_load(Unit *u) {
         return socket_verify(s);
 }
 
-static SocketPeer *socket_peer_new(void) {
+static SocketPeer *socket_peer_dup(const SocketPeer *q) {
         SocketPeer *p;
 
+        assert(q);
+
         p = new(SocketPeer, 1);
         if (!p)
                 return NULL;
 
         *p = (SocketPeer) {
                 .n_ref = 1,
+                .peer = q->peer,
+                .peer_salen = q->peer_salen,
+                .peer_cred = q->peer_cred,
         };
+
         return p;
 }
 
@@ -492,8 +503,9 @@ DEFINE_TRIVIAL_REF_UNREF_FUNC(SocketPeer, socket_peer, socket_peer_free);
 
 int socket_acquire_peer(Socket *s, int fd, SocketPeer **ret) {
         _cleanup_(socket_peer_unrefp) SocketPeer *remote = NULL;
-        SocketPeer sa = {
+        SocketPeer key = {
                 .peer_salen = sizeof(union sockaddr_union),
+                .peer_cred = UCRED_INVALID,
         }, *i;
         int r;
 
@@ -501,27 +513,36 @@ int socket_acquire_peer(Socket *s, int fd, SocketPeer **ret) {
         assert(s);
         assert(ret);
 
-        if (getpeername(fd, &sa.peer.sa, &sa.peer_salen) < 0)
+        if (getpeername(fd, &key.peer.sa, &key.peer_salen) < 0)
                 return log_unit_error_errno(UNIT(s), errno, "getpeername() failed: %m");
 
-        if (!IN_SET(sa.peer.sa.sa_family, AF_INET, AF_INET6, AF_VSOCK)) {
+        switch (key.peer.sa.sa_family) {
+        case AF_INET:
+        case AF_INET6:
+        case AF_VSOCK:
+                break;
+
+        case AF_UNIX:
+                r = getpeercred(fd, &key.peer_cred);
+                if (r < 0)
+                        return log_unit_error_errno(UNIT(s), r, "Failed to get peer credentials of socket: %m");
+                break;
+
+        default:
                 *ret = NULL;
                 return 0;
         }
 
-        i = set_get(s->peers_by_address, &sa);
+        i = set_get(s->peers_by_address, &key);
         if (i) {
                 *ret = socket_peer_ref(i);
                 return 1;
         }
 
-        remote = socket_peer_new();
+        remote = socket_peer_dup(&key);
         if (!remote)
                 return log_oom();
 
-        remote->peer = sa.peer;
-        remote->peer_salen = sa.peer_salen;
-
         r = set_ensure_put(&s->peers_by_address, &peer_address_hash_ops, remote);
         if (r < 0)
                 return log_unit_error_errno(UNIT(s), r, "Failed to insert peer info into hash table: %m");
@@ -2300,7 +2321,10 @@ static void socket_enter_running(Socket *s, int cfd_in) {
                         if (r > 0 && p->n_ref > s->max_connections_per_source) {
                                 _cleanup_free_ char *t = NULL;
 
-                                (void) sockaddr_pretty(&p->peer.sa, p->peer_salen, true, false, &t);
+                                if (p->peer.sa.sa_family == AF_UNIX)
+                                        (void) asprintf(&t, "UID " UID_FMT, p->peer_cred.uid);
+                                else
+                                        (void) sockaddr_pretty(&p->peer.sa, p->peer_salen, /* translate_ipv6= */ true, /* include_port= */ false, &t);
 
                                 log_unit_warning(UNIT(s),
                                                  "Too many incoming connections (%u) from source %s, dropping connection.",