(
self,
message: dns.DNSMessage,
)
| 114 | ) |
| 115 | |
| 116 | async def resolve( |
| 117 | self, |
| 118 | message: dns.DNSMessage, |
| 119 | ) -> dns.DNSMessage: |
| 120 | q = message.question |
| 121 | assert q |
| 122 | try: |
| 123 | if q.type == dns.types.A: |
| 124 | ip_addrs = await self.resolver().lookup_ipv4(q.name) |
| 125 | else: |
| 126 | ip_addrs = await self.resolver().lookup_ipv6(q.name) |
| 127 | except socket.gaierror as e: |
| 128 | match e.args[0]: |
| 129 | case socket.EAI_NONAME: |
| 130 | return message.fail(dns.response_codes.NXDOMAIN) |
| 131 | case socket.EAI_NODATA: |
| 132 | ip_addrs = [] |
| 133 | case _: |
| 134 | return message.fail(dns.response_codes.SERVFAIL) |
| 135 | |
| 136 | return message.succeed( |
| 137 | [ |
| 138 | dns.ResourceRecord( |
| 139 | name=q.name, |
| 140 | type=q.type, |
| 141 | class_=q.class_, |
| 142 | ttl=dns.ResourceRecord.DEFAULT_TTL, |
| 143 | data=ipaddress.ip_address(ip).packed, |
| 144 | ) |
| 145 | for ip in ip_addrs |
| 146 | ] |
| 147 | ) |
| 148 | |
| 149 | |
| 150 | class Resolver(Protocol): |
no test coverage detected