@@ -439,6 +439,17 @@ def new_respond_socket( |
439 | 439 | return respond_socket |
440 | 440 |
|
441 | 441 |
|
| 442 | +def _entry_ip_version(entry: str | int | tuple[tuple[str, int, int], int]) -> int: |
| 443 | + """Return 4 or 6 for the IP version implied by an interface / multicast entry. |
| 444 | +
|
| 445 | + ``int`` (interface index) and ``tuple`` (ifaddr IPv6 adapter tuple) entries |
| 446 | + are IPv6 by construction; ``str`` entries are parsed as IP addresses. |
| 447 | + """ |
| 448 | + if isinstance(entry, str): |
| 449 | + return ipaddress.ip_address(entry).version |
| 450 | + return 6 |
| 451 | + |
| 452 | + |
442 | 453 | def create_sockets( |
443 | 454 | interfaces: InterfacesType = InterfaceChoice.All, |
444 | 455 | unicast: bool = False, |
@@ -461,16 +472,10 @@ def create_sockets( |
461 | 472 | # Reject IP-version-incompatible entries up front so callers get a clear |
462 | 473 | # error instead of a confusing adapter-lookup or socket-syscall failure. |
463 | 474 | if multicast_addresses: |
464 | | - if ip_version == IPVersion.V4Only: |
465 | | - for entry in multicast_addresses: |
466 | | - if isinstance(entry, (int, tuple)) or ( |
467 | | - isinstance(entry, str) and ipaddress.ip_address(entry).version == 6 |
468 | | - ): |
469 | | - raise ValueError("multicast_addresses contains IPv6 entries but ip_version is V4Only") |
470 | | - elif ip_version == IPVersion.V6Only: |
471 | | - for entry in multicast_addresses: |
472 | | - if isinstance(entry, str) and ipaddress.ip_address(entry).version == 4: |
473 | | - raise ValueError("multicast_addresses contains IPv4 entries but ip_version is V6Only") |
| 475 | + if ip_version == IPVersion.V4Only and any(_entry_ip_version(e) == 6 for e in multicast_addresses): |
| 476 | + raise ValueError("multicast_addresses contains IPv6 entries but ip_version is V4Only") |
| 477 | + if ip_version == IPVersion.V6Only and any(_entry_ip_version(e) == 4 for e in multicast_addresses): |
| 478 | + raise ValueError("multicast_addresses contains IPv4 entries but ip_version is V6Only") |
474 | 479 |
|
475 | 480 | if unicast: |
476 | 481 | listen_socket = None |
@@ -534,22 +539,13 @@ def autodetect_ip_version( |
534 | 539 | multicast_addresses: Sequence[str | int | tuple[tuple[str, int, int], int]] | None = None, |
535 | 540 | ) -> IPVersion: |
536 | 541 | """Auto detect the IP version when it is not provided.""" |
537 | | - has_v6 = False |
538 | | - has_v4 = False |
| 542 | + entries: list[str | int | tuple[tuple[str, int, int], int]] = [] |
539 | 543 | if isinstance(interfaces, list): |
540 | | - has_v6 = any( |
541 | | - isinstance(i, int) or (isinstance(i, str) and ipaddress.ip_address(i).version == 6) |
542 | | - for i in interfaces |
543 | | - ) |
544 | | - has_v4 = any(isinstance(i, str) and ipaddress.ip_address(i).version == 4 for i in interfaces) |
| 544 | + entries.extend(interfaces) |
545 | 545 | if multicast_addresses: |
546 | | - has_v6 = has_v6 or any( |
547 | | - isinstance(i, (int, tuple)) or (isinstance(i, str) and ipaddress.ip_address(i).version == 6) |
548 | | - for i in multicast_addresses |
549 | | - ) |
550 | | - has_v4 = has_v4 or any( |
551 | | - isinstance(i, str) and ipaddress.ip_address(i).version == 4 for i in multicast_addresses |
552 | | - ) |
| 546 | + entries.extend(multicast_addresses) |
| 547 | + has_v6 = any(_entry_ip_version(e) == 6 for e in entries) |
| 548 | + has_v4 = any(_entry_ip_version(e) == 4 for e in entries) |
553 | 549 | if has_v4 and has_v6: |
554 | 550 | return IPVersion.All |
555 | 551 | if has_v6: |
|