feat(mld): MCAST_JOIN_SOURCE_GROUP family for IPv6 SSM (R4 P2, part 2)

Adds the protocol-independent IPv6 multicast source-filter socket options
— the second half of R4 P2, completing the socket surface for IPv6
source-specific multicast.

New 'McastOption' IntEnum + bare aliases (enums.md §2.2 stdlib-parity
form, re-exported from 'pytcp.socket'): MCAST_JOIN_SOURCE_GROUP (46) /
MCAST_LEAVE_SOURCE_GROUP (47) / MCAST_BLOCK_SOURCE (43) /
MCAST_UNBLOCK_SOURCE (44), wired at IPPROTO_IPV6 level. There is NO
'IPV6_ADD_SOURCE_MEMBERSHIP' — Linux drives IPv6 SSM through the
protocol-independent 'MCAST_*' family with a 'group_source_req' struct,
not the IPv4-style 'ip_mreq_source'.

A module-level '_parse_group_source_req' decodes the 264-byte glibc
'struct group_source_req' (uint32 gsr_interface at 0; sockaddr_storage
gsr_group at 8 → group addr bytes[16:32]; gsr_source at 136 → source addr
bytes[144:160]), validating the total length and that each embedded
sockaddr's ss_family is AF_INET6 (the OS-level value 10, the same
'error_queue' packs — distinct from PyTCP's internal 'AddressFamily.INET6'
enum). '_apply_ip6_source_op' computes the new per-socket filter with the
Linux 'ip6_mc_source' errno surface (EINVAL on an INCLUDE/EXCLUDE
filter-mode conflict or BLOCK/UNBLOCK without a prior any-source join,
EADDRNOTAVAIL on dropping/unblocking a source not held) — the IPv6
analogue of the v4 '_apply_source_op'. '_ipproto_ipv6_source_membership'
performs the get/compute/store as one atomic RMW under
'_lock__ip6_source_filters' and pushes the result to the interface merge
via 'stack.membership6'.

Tests-first: 'test__socket__ipv6_source_membership.py' drives all four
options end-to-end through the socket (INCLUDE join materializes
INCLUDE{src}, source accumulation, leave-last-source leaves the group,
BLOCK/UNBLOCK on an any-source membership, the mode-conflict + stale-source
errno paths, short / wrong-family struct rejection, and close-time
release). The MLDv2 source records on the wire (P4) and the RX
source-delivery gate (P5) remain before the RFC 3810 §4.2/§5.2 adherence
flip.

Lint clean, 13527 passing, 0 skipped.

Reference: RFC 3810 §4.1 (per-socket filter mode + source list).
Reference: RFC 3678 §4 (MCAST_JOIN_SOURCE_GROUP / group_source_req).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ccie18643

docs/refactor/host_refinements_backlog.md

@@ -90,7 +90,21 @@ until the user says "push".
   branch is a `# P4:` placeholder). Tests-first:
   `test__icmp6__mld__source_filter_model.py` (7 wire-driven model tests) +
   harness / thread-safety / ND-seed updates. RFC 3810 §4.2/§5.2 adherence
-  flip waits for P4/P5. See the R4 section for the remaining P2/P4/P5 map.
+  flip waits for P4/P5. See the R4 section for the remaining P4/P5 map.
+- [x] **IPv6 SSM socket surface (R4 P2)** — the socket-side source-filter
+  API, in two commits. `stack/membership6.py` (`Membership6Api`, the MLDv2
+  analogue of `MembershipApi`) registered as `stack.membership6`;
+  `IPV6_JOIN_GROUP` / `IPV6_LEAVE_GROUP` migrated onto it with per-socket
+  `_ip6_source_filters` refcounting (fixes the leave-removes-for-all bug)
+  + close-time release; and the protocol-independent
+  `MCAST_JOIN_SOURCE_GROUP` / `MCAST_LEAVE_SOURCE_GROUP` /
+  `MCAST_BLOCK_SOURCE` / `MCAST_UNBLOCK_SOURCE` family (`McastOption` enum +
+  bare aliases, `group_source_req` parser, `_apply_ip6_source_op`). There
+  is NO `IPV6_ADD_SOURCE_MEMBERSHIP` — Linux uses the `MCAST_*` family for
+  IPv6 SSM. Tests: `test__socket__ipv6_source_membership.py` (11) +
+  `test__icmp6__mld__membership6_api.py` (8) + the join_group refcount /
+  close tests. P4 (MLDv2 source records on the wire) + P5 (RX source-
+  delivery gate) remain before the §4.2/§5.2 adherence flip.
 
 **The canonical SO_RCVBUF guard pattern** (mirror for any new datagram socket):
 
@@ -187,10 +201,28 @@ self._signal_readable()
     v4/v6 multicast are already separate paths in the handler and the
     MLD/IGMP TX). P3's `_ip6_multicast_filters` dict + merge live on the
     handler next to the v4 ones.
-  - P2: new opts `IPV6_ADD_SOURCE_MEMBERSHIP` / `IPV6_DROP_SOURCE_MEMBERSHIP`
-    (and/or the protocol-independent `MCAST_JOIN_SOURCE_GROUP` family) as
-    `IpV6Option` enum members + bare aliases (see `.claude/rules/enums.md`
-    §2.2). Wire the setsockopt cases.
+  - **P2 — SHIPPED:** the socket-side source-filter surface + IPv6
+    membership API, in two commits. (1) `stack/membership6.py`
+    (`Membership6Api`, the MLDv2 analogue of `MembershipApi`: join / leave
+    / set_socket_filter / clear_socket_filter / list_memberships →
+    `mc6_*`; ff02::1 leave-refusal; no group-count cap — Linux has no IPv6
+    `igmp_max_memberships`), registered as `stack.membership6`. The
+    `IPV6_JOIN_GROUP` / `IPV6_LEAVE_GROUP` path migrated onto it with
+    per-socket `_ip6_source_filters` refcounting (EXCLUDE{} any-source),
+    fixing the documented leave-removes-for-all bug; close/GC releases
+    held memberships (`_release_ip6_memberships`). (2) the
+    protocol-independent `MCAST_JOIN_SOURCE_GROUP` /
+    `MCAST_LEAVE_SOURCE_GROUP` / `MCAST_BLOCK_SOURCE` /
+    `MCAST_UNBLOCK_SOURCE` family as a `McastOption` IntEnum + bare aliases
+    (enums.md §2.2; re-exported from `pytcp.socket`), with a
+    `group_source_req` parser (`_parse_group_source_req`, ss_family ==
+    AF_INET6 = 10) and `_apply_ip6_source_op` (the INCLUDE/EXCLUDE
+    mode-conflict + EADDRNOTAVAIL errno surface mirroring the v4
+    `_apply_source_op`). There is NO `IPV6_ADD_SOURCE_MEMBERSHIP` — Linux
+    uses the protocol-independent `MCAST_*` family for IPv6 SSM.
+    Tests-first: `test__socket__ipv6_source_membership.py` (11) +
+    `test__icmp6__mld__membership6_api.py` (8) + the refcount / close tests
+    in `test__socket__ipv6_join_group.py`.
   - **P3 — SHIPPED:** the handler `mc6_*` reception-state core, a
     faithful mirror of the fully-evolved v4 machinery. Added
     `_Ip6GroupMembership`, the `_ip6_multicast_filters` dict as the
@@ -225,14 +257,17 @@ self._signal_readable()
 #### R4 — resumable implementation map (derived 2026-07-19 by reading the v4 machinery)
 
 The v4 source-filter machinery to mirror, with exact locations, so P2-P5
-can resume without re-deriving. **Status: P3 (the handler core below) is
-SHIPPED** — the `_Ip6GroupMembership` / `_ip6_multicast_filters` /
-`_ip6_multicast_refs` / `mc6_*` / `_ip6_multicast_filter_for` machinery all
-landed under `_lock__multicast`, and `assign/remove_ip6_multicast` now
-materialize the filter map. **Remaining: P2** (Membership API + Socket
-`MCAST_*` setsockopt + `group_source_req`), **P4** (`_send_mld_state_change`
-MLDv2 source records, replacing the `# P4:` placeholder in `_mc6_recompute`),
-**P5** (RX `Ip6MulticastFilter.allows` gate for UDP + RAW).
+can resume without re-deriving. **Status: P2 + P3 SHIPPED** — the handler
+core (`_Ip6GroupMembership` / `_ip6_multicast_filters` / `_ip6_multicast_refs`
+/ `mc6_*` / `_ip6_multicast_filter_for`, all under `_lock__multicast`) AND
+the socket surface (`stack.membership6`, refcounted `IPV6_JOIN_GROUP`, the
+`MCAST_*_SOURCE_*` family with `group_source_req`, per-socket
+`_ip6_source_filters`, close-time release) are done. **Remaining: P4**
+(`_send_mld_state_change` MLDv2 source records — ALLOW / BLOCK / CHANGE_TO_*
+— replacing the `# P4:` current-state-Report placeholder in `_mc6_recompute`
+and the join/leave edges), **P5** (RX `Ip6MulticastFilter.allows` source-
+delivery gate for UDP + RAW). After P4/P5 flip the RFC 3810 §4.2 / §5.2
+adherence rows.
 
 - **Handler (`runtime/packet_handler/__init__.py`) — the deep, no-GIL-locked
   core, duplicated across the L2 and L3 handler classes — SHIPPED (P3):**

packages/pytcp/pytcp/runtime/socket/__init__.py

@@ -256,6 +256,29 @@ class IpV6Option(IntEnum):
 IPV6_TCLASS = IpV6Option.IPV6_TCLASS
 
 
+class McastOption(IntEnum):
+    """
+    Protocol-independent multicast source-filter setsockopt 'optname'
+    values (RFC 3678 / RFC 3810 §4.1), usable at both IPPROTO_IP and
+    IPPROTO_IPV6 level with a 'group_source_req' struct. Linux numbers
+    from <linux/in.h>; matches Python stdlib 'socket.MCAST_*' on
+    platforms that expose them. PyTCP wires these at IPPROTO_IPV6 for
+    IPv6 source-specific multicast; IPv4 uses the older 'IP_*_SOURCE_*'
+    'ip_mreq_source' options.
+    """
+
+    MCAST_BLOCK_SOURCE = 43  # group_source_req: block a source on an EXCLUDE-mode group
+    MCAST_UNBLOCK_SOURCE = 44  # group_source_req: unblock a source on an EXCLUDE-mode group
+    MCAST_JOIN_SOURCE_GROUP = 46  # group_source_req: join a group in INCLUDE mode for a source
+    MCAST_LEAVE_SOURCE_GROUP = 47  # group_source_req: drop a source from an INCLUDE-mode group
+
+
+MCAST_BLOCK_SOURCE = McastOption.MCAST_BLOCK_SOURCE
+MCAST_UNBLOCK_SOURCE = McastOption.MCAST_UNBLOCK_SOURCE
+MCAST_JOIN_SOURCE_GROUP = McastOption.MCAST_JOIN_SOURCE_GROUP
+MCAST_LEAVE_SOURCE_GROUP = McastOption.MCAST_LEAVE_SOURCE_GROUP
+
+
 class UdpOption(IntEnum):
     """
     SOL_UDP-level setsockopt 'optname' values (Linux numbers
@@ -366,6 +389,42 @@ def _resolve_membership_ifindex(interface_address: Ip4Address, /) -> int | None:
     return None
 
 
+# 'struct group_source_req' (glibc, native layout): uint32 gsr_interface
+# at offset 0; sockaddr_storage gsr_group at offset 8 (sockaddr_in6 ->
+# ss_family at +0, sin6_addr at +8, so the group address is bytes[16:32]);
+# sockaddr_storage gsr_source at offset 136 (source address bytes[144:160]).
+GROUP_SOURCE_REQ__LEN: int = 264
+# Linux OS-level AF_INET6 as it appears in a sockaddr wire struct — the
+# same value 'error_queue._AF_INET6' packs, distinct from PyTCP's
+# internal 'AddressFamily.INET6' socket-family enum value.
+_AF_INET6__SOCKADDR: int = 10
+
+
+def _parse_group_source_req(gsr: bytes, /) -> tuple[int, Ip6Address, Ip6Address]:
+    """
+    Parse a Linux 'struct group_source_req' (RFC 3678) into
+    (ifindex, group, source). Validates the total length and that each
+    embedded sockaddr's ss_family is AF_INET6. Raises 'OSError(EINVAL)'
+    on a short buffer or a non-AF_INET6 sockaddr family.
+    """
+
+    if len(gsr) < GROUP_SOURCE_REQ__LEN:
+        raise OSError(
+            errno.EINVAL,
+            f"group_source_req must be at least {GROUP_SOURCE_REQ__LEN} bytes, got {len(gsr)}",
+        )
+
+    ifindex = int.from_bytes(gsr[0:4], sys.byteorder)
+    group_family = int.from_bytes(gsr[8:10], sys.byteorder)
+    source_family = int.from_bytes(gsr[136:138], sys.byteorder)
+    if group_family != _AF_INET6__SOCKADDR or source_family != _AF_INET6__SOCKADDR:
+        raise OSError(errno.EINVAL, "group_source_req sockaddr family must be AF_INET6")
+
+    group = Ip6Address(gsr[16:32])
+    source = Ip6Address(gsr[144:160])
+    return ifindex, group, source
+
+
 class ShutdownHow(IntEnum):
     """
     BSD-socket 'shutdown(how)' values per POSIX (Linux-
@@ -1044,6 +1103,20 @@ def _ipproto_ipv6_setsockopt(self, optname: int, value: int | bytes, /) -> bool:
                     )
                 self._ipproto_ipv6_membership(optname, bytes(value))
                 return True
+            case _ if optname in (
+                MCAST_JOIN_SOURCE_GROUP,
+                MCAST_LEAVE_SOURCE_GROUP,
+                MCAST_BLOCK_SOURCE,
+                MCAST_UNBLOCK_SOURCE,
+            ):
+                if not isinstance(value, (bytes, bytearray, memoryview)):
+                    raise OSError(
+                        errno.EINVAL,
+                        f"MCAST source-membership value must be a group_source_req bytes object, "
+                        f"got {type(value).__name__}",
+                    )
+                self._ipproto_ipv6_source_membership(optname, bytes(value))
+                return True
         return False
 
     def _resolve_ipv6_membership_interface(self, mreq_ifindex: int, /) -> int:
@@ -1114,6 +1187,100 @@ def _ipproto_ipv6_membership(self, optname: int, mreq: bytes, /) -> None:
         except ValueError as error:
             raise OSError(errno.EINVAL, str(error)) from error
 
+    def _ipproto_ipv6_source_membership(self, optname: int, gsr: bytes, /) -> None:
+        """
+        Apply the IPv6 protocol-independent source-filter socket options
+        (RFC 3678 / RFC 3810 §4.1) by parsing the 'group_source_req'
+        structure and mutating this socket's per-(ifindex, group) filter,
+        then pushing the result to the interface merge via the IPv6
+        membership API:
+
+        - MCAST_JOIN_SOURCE_GROUP: INCLUDE-mode; add the source.
+        - MCAST_LEAVE_SOURCE_GROUP: remove the source from the INCLUDE
+          list (leave the group when it empties).
+        - MCAST_BLOCK_SOURCE: EXCLUDE-mode; add the blocked source.
+        - MCAST_UNBLOCK_SOURCE: remove a blocked source from the EXCLUDE
+          list.
+
+        A mode conflict (an INCLUDE op on an EXCLUDE membership or vice
+        versa, and BLOCK / UNBLOCK with no prior any-source join) raises
+        EINVAL, mirroring Linux 'ip6_mc_source'. The IPv6 analogue of
+        '_ipproto_ip_source_membership'.
+        """
+
+        import pytcp.stack as _stack
+
+        mreq_ifindex, group, source = _parse_group_source_req(gsr)
+        if not group.is_multicast:
+            raise OSError(errno.EINVAL, f"{group} is not a multicast group address")
+        ifindex = self._resolve_ipv6_membership_interface(mreq_ifindex)
+
+        key = (ifindex, group)
+        api = _stack.membership6.interface(ifindex)
+        try:
+            with self._lock__ip6_source_filters:
+                # Compute the new socket filter (or None to leave the
+                # group) inside the lock so the get/compute/store is one
+                # atomic RMW and a mode-conflict OSError aborts cleanly.
+                new_filter = self._apply_ip6_source_op(optname, self._ip6_source_filters.get(key), source)
+                if new_filter is None:
+                    api.clear_socket_filter(group=group, token=id(self))
+                    self._ip6_source_filters.pop(key, None)
+                else:
+                    api.set_socket_filter(group=group, token=id(self), source_filter=new_filter)
+                    self._ip6_source_filters[key] = new_filter
+        except ValueError as error:
+            raise OSError(errno.EINVAL, str(error)) from error
+
+    @staticmethod
+    def _apply_ip6_source_op(
+        optname: int,
+        current: Ip6MulticastFilter | None,
+        source: Ip6Address,
+        /,
+    ) -> Ip6MulticastFilter | None:
+        """
+        Compute the new per-socket IPv6 source filter for a source-filter
+        socket option (RFC 3810 §4.1 / RFC 3678), or 'None' when the
+        operation leaves the group. Raises 'OSError' with the Linux
+        'ip6_mc_source' errno for an invalid transition: EINVAL on a
+        filter-mode conflict, EADDRNOTAVAIL on dropping / unblocking a
+        source the socket does not hold. Adding a source already present
+        is idempotent. The IPv6 analogue of '_apply_source_op'.
+        """
+
+        include = Ip6MulticastFilterMode.INCLUDE
+        exclude = Ip6MulticastFilterMode.EXCLUDE
+
+        match optname:
+            case _ if optname == MCAST_JOIN_SOURCE_GROUP:
+                if current is None:
+                    return Ip6MulticastFilter(include, frozenset({source}))
+                if current.mode is exclude:
+                    raise OSError(errno.EINVAL, "MCAST_JOIN_SOURCE_GROUP on an EXCLUDE-mode group")
+                return Ip6MulticastFilter(include, current.sources | {source})
+            case _ if optname == MCAST_LEAVE_SOURCE_GROUP:
+                if current is None:
+                    raise OSError(errno.EADDRNOTAVAIL, "Socket is not a member of the group")
+                if current.mode is exclude:
+                    raise OSError(errno.EINVAL, "MCAST_LEAVE_SOURCE_GROUP on an EXCLUDE-mode group")
+                if source not in current.sources:
+                    raise OSError(errno.EADDRNOTAVAIL, f"Source {source} is not in the include list")
+                remaining = current.sources - {source}
+                return Ip6MulticastFilter(include, remaining) if remaining else None
+            case _ if optname == MCAST_BLOCK_SOURCE:
+                if current is None or current.mode is include:
+                    raise OSError(errno.EINVAL, "MCAST_BLOCK_SOURCE requires an EXCLUDE-mode (any-source) membership")
+                return Ip6MulticastFilter(exclude, current.sources | {source})
+            case _ if optname == MCAST_UNBLOCK_SOURCE:
+                if current is None or current.mode is include:
+                    raise OSError(errno.EINVAL, "MCAST_UNBLOCK_SOURCE requires an EXCLUDE-mode (any-source) membership")
+                if source not in current.sources:
+                    raise OSError(errno.EADDRNOTAVAIL, f"Source {source} is not blocked")
+                return Ip6MulticastFilter(exclude, current.sources - {source})
+
+        raise OSError(errno.EINVAL, f"Unsupported source-membership option {optname}")
+
     def _ipproto_ipv6_getsockopt(self, optname: int, /) -> int | None:
         """
         Get an IPPROTO_IPV6-level option's stored value, or 'None' if

packages/pytcp/pytcp/socket/__init__.py

@@ -91,6 +91,10 @@
     IPV6_TCLASS,
     IPV6_UNICAST_HOPS,
     IPV6_V6ONLY,
+    MCAST_BLOCK_SOURCE,
+    MCAST_JOIN_SOURCE_GROUP,
+    MCAST_LEAVE_SOURCE_GROUP,
+    MCAST_UNBLOCK_SOURCE,
     MSG_ERRQUEUE,
     MSG_OOB,
     PACKET_BROADCAST,
@@ -208,6 +212,10 @@
     "IP_TOS",
     "IP_TTL",
     "IP_UNBLOCK_SOURCE",
+    "MCAST_BLOCK_SOURCE",
+    "MCAST_JOIN_SOURCE_GROUP",
+    "MCAST_LEAVE_SOURCE_GROUP",
+    "MCAST_UNBLOCK_SOURCE",
     "MSG_ERRQUEUE",
     "MSG_OOB",
     "PACKET_BROADCAST",