from ipaddress import IPv4Address, IPv6Address, _BaseAddress

from typing import TYPE_CHECKING, Dict, List, Optional, Set, Union, cast

from .._cache import DNSCache

from .._dns import (

DNSAddress,

DNSNsec,

DNSPointer,

DNSQuestion,

DNSQuestionType,

DNSRecord,

DNSService,

DNSText,

)

from .._exceptions import BadTypeInNameException

from .._history import QuestionHistory

from .._logger import log

from .._protocol.outgoing import DNSOutgoing

from .._record_update import RecordUpdate

_CLASS_IN_UNIQUE,

_DNS_HOST_TTL,

_DNS_OTHER_TTL,

_DUPLICATE_QUESTION_INTERVAL,

_FLAGS_QR_QUERY,

_LISTENER_TIME,

_MDNS_PORT,

bytes_ = bytes

float_ = float

int_ = int

str_ = str

DNS_QUESTION_TYPE_QU = DNSQuestionType.QU

DNS_QUESTION_TYPE_QM = DNSQuestionType.QM

QU_QUESTION = DNSQuestionType.QU

QM_QUESTION = DNSQuestionType.QM

randint = random.randint

if TYPE_CHECKING:

from .._core import Zeroconf

)

)

def _get_initial_delay(self) -> float_:

return _LISTENER_TIME

def _get_random_delay(self) -> int_:

return randint(*_AVOID_SYNC_DELAY_RANDOM_INTERVAL)

async def async_request(

self,

zc: 'Zeroconf',

assert zc.loop is not None

first_request = True

delay = _LISTENER_TIME

delay = self._get_initial_delay()

next_ = now

last = now + timeout

try:

if last <= now:

return False

if next_ <= now:

out = self._generate_request_query(

zc,

now,

question_type or DNS_QUESTION_TYPE_QU if first_request else DNS_QUESTION_TYPE_QM,

)

this_question_type = question_type or QU_QUESTION if first_request else QM_QUESTION

out = self._generate_request_query(zc, now, this_question_type)

first_request = False

if not out.questions:

return self._load_from_cache(zc, now)

zc.async_send(out, addr, port)

if out.questions:

# All questions may have been suppressed

# by the question history, so nothing to send,

# but keep waiting for answers in case another

# client on the network is asking the same

# question or they have not arrived yet.

zc.async_send(out, addr, port)

next_ = now + delay

delay *= 2

next_ += random.randint(*_AVOID_SYNC_DELAY_RANDOM_INTERVAL)

next_ += self._get_random_delay()

if this_question_type is QM_QUESTION and delay < _DUPLICATE_QUESTION_INTERVAL:

# If we just asked a QM question, we need to

# wait at least the duplicate question interval

# before asking another QM question otherwise

# its likely to be suppressed by the question

# history of the remote responder.

delay = _DUPLICATE_QUESTION_INTERVAL

await self.async_wait(min(next_, last) - now, zc.loop)

now = current_time_millis()

return True

def _add_question_with_known_answers(

self,

out: DNSOutgoing,

qu_question: bool,

question_history: QuestionHistory,

cache: DNSCache,

now: float_,

name: str_,

type_: int_,

class_: int_,

skip_if_known_answers: bool,

) -> None:

"""Add a question with known answers if its not suppressed."""

known_answers = {

answer for answer in cache.get_all_by_details(name, type_, class_) if not answer.is_stale(now)

}

if skip_if_known_answers and known_answers:

return

question = DNSQuestion(name, type_, class_)

if qu_question:

question.unicast = True

elif question_history.suppresses(question, now, known_answers):

return

else:

question_history.add_question_at_time(question, now, known_answers)

out.add_question(question)

for answer in known_answers:

out.add_answer_at_time(answer, now)

def _generate_request_query(

self, zc: 'Zeroconf', now: float_, question_type: DNSQuestionType

) -> DNSOutgoing:

"""Generate the request query."""

out = DNSOutgoing(_FLAGS_QR_QUERY)

name = self._name

server_or_name = self.server or name

server = self.server or name

cache = zc.cache

out.add_question_or_one_cache(cache, now, name, _TYPE_SRV, _CLASS_IN)

out.add_question_or_one_cache(cache, now, name, _TYPE_TXT, _CLASS_IN)

out.add_question_or_all_cache(cache, now, server_or_name, _TYPE_A, _CLASS_IN)

out.add_question_or_all_cache(cache, now, server_or_name, _TYPE_AAAA, _CLASS_IN)

if question_type is DNS_QUESTION_TYPE_QU:

for question in out.questions:

question.unicast = True

history = zc.question_history

qu_question = question_type is QU_QUESTION

self._add_question_with_known_answers(

out, qu_question, history, cache, now, name, _TYPE_SRV, _CLASS_IN, True

)

self._add_question_with_known_answers(

out, qu_question, history, cache, now, name, _TYPE_TXT, _CLASS_IN, True

)

self._add_question_with_known_answers(

out, qu_question, history, cache, now, server, _TYPE_A, _CLASS_IN, False

)

self._add_question_with_known_answers(

out, qu_question, history, cache, now, server, _TYPE_AAAA, _CLASS_IN, False

)

return out

def __repr__(self) -> str: