(
q: dns.message.Message,
where: str,
url: str,
timeout: float | None = None,
port: int = 443,
source: str | None = None,
source_port: int = 0,
one_rr_per_rrset: bool = False,
ignore_trailing: bool = False,
verify: bool | str | ssl.SSLContext = True,
backend: dns.asyncbackend.Backend | None = None,
post: bool = True,
connection: dns.quic.AsyncQuicConnection | None = None,
)
| 704 | |
| 705 | |
| 706 | async def _http3( |
| 707 | q: dns.message.Message, |
| 708 | where: str, |
| 709 | url: str, |
| 710 | timeout: float | None = None, |
| 711 | port: int = 443, |
| 712 | source: str | None = None, |
| 713 | source_port: int = 0, |
| 714 | one_rr_per_rrset: bool = False, |
| 715 | ignore_trailing: bool = False, |
| 716 | verify: bool | str | ssl.SSLContext = True, |
| 717 | backend: dns.asyncbackend.Backend | None = None, |
| 718 | post: bool = True, |
| 719 | connection: dns.quic.AsyncQuicConnection | None = None, |
| 720 | ) -> dns.message.Message: |
| 721 | if not dns.quic.have_quic: |
| 722 | raise NoDOH("DNS-over-HTTP3 is not available.") # pragma: no cover |
| 723 | |
| 724 | url_parts = urllib.parse.urlparse(url) |
| 725 | hostname = url_parts.hostname |
| 726 | assert hostname is not None |
| 727 | if url_parts.port is not None: |
| 728 | port = url_parts.port |
| 729 | |
| 730 | q.id = 0 |
| 731 | wire = q.to_wire() |
| 732 | the_connection: dns.quic.AsyncQuicConnection |
| 733 | if connection: |
| 734 | cfactory = dns.quic.null_factory |
| 735 | mfactory = dns.quic.null_factory |
| 736 | else: |
| 737 | (cfactory, mfactory) = dns.quic.factories_for_backend(backend) |
| 738 | |
| 739 | async with cfactory() as context: |
| 740 | async with mfactory( |
| 741 | context, verify_mode=verify, server_name=hostname, h3=True |
| 742 | ) as the_manager: |
| 743 | if connection: |
| 744 | the_connection = connection |
| 745 | else: |
| 746 | the_connection = the_manager.connect( # pyright: ignore |
| 747 | where, port, source, source_port |
| 748 | ) |
| 749 | (start, expiration) = _compute_times(timeout) |
| 750 | stream = await the_connection.make_stream(timeout) # pyright: ignore |
| 751 | async with stream: |
| 752 | # note that send_h3() does not need await |
| 753 | stream.send_h3(url, wire, post) |
| 754 | wire = await stream.receive(_remaining(expiration)) |
| 755 | _check_status(stream.headers(), where, wire) |
| 756 | finish = time.time() |
| 757 | r = dns.message.from_wire( |
| 758 | wire, |
| 759 | keyring=q.keyring, |
| 760 | request_mac=q.request_mac, |
| 761 | one_rr_per_rrset=one_rr_per_rrset, |
| 762 | ignore_trailing=ignore_trailing, |
| 763 | ) |
no test coverage detected
searching dependent graphs…