Refactor ControllableAsyncDnsServer setup

When this class was introduced, the constructor of its base class had no
parameters. This was changed in the meantime and these parameters were
not accessible by users of the subclass.

Don't override the constructor.
Move command setup to methods.
Move subclass-specific storage to cached properties.
Take instances of Command instead of the classes themselves for
symmetry with install_response_handler.

(cherry picked from commit a0970f3d04)
This commit is contained in:
Štěpán Balážik 2025-10-30 13:41:23 +01:00 committed by Štěpán Balážik (GitLab job 6645047)
parent 5c897cca32
commit a2c444760d
6 changed files with 33 additions and 22 deletions

View file

@ -473,7 +473,8 @@ class ChainResponseHandler(DomainHandler):
def main() -> None:
server = ControllableAsyncDnsServer(commands=[ChainSetupCommand])
server = ControllableAsyncDnsServer()
server.install_control_command(ChainSetupCommand())
server.run()

View file

@ -39,7 +39,8 @@ class MaybeDelayedAddressAnswerHandler(ResponseHandler):
def main() -> None:
server = ControllableAsyncDnsServer([ToggleResponsesCommand])
server = ControllableAsyncDnsServer()
server.install_control_command(ToggleResponsesCommand())
server.install_response_handler(MaybeDelayedAddressAnswerHandler())
server.run()

View file

@ -49,7 +49,8 @@ class ExtraAnswersHandler(DomainHandler):
def main() -> None:
server = ControllableAsyncDnsServer(commands=[ToggleResponsesCommand])
server = ControllableAsyncDnsServer()
server.install_control_command(ToggleResponsesCommand())
server.install_response_handler(ExtraAnswersHandler())
server.run()

View file

@ -72,7 +72,8 @@ class ChaseDsHandler(ResponseHandler):
def main() -> None:
server = ControllableAsyncDnsServer([ToggleResponsesCommand])
server = ControllableAsyncDnsServer()
server.install_control_command(ToggleResponsesCommand())
server.install_response_handler(ChaseDsHandler())
server.run()

View file

@ -21,7 +21,6 @@ from typing import (
List,
Optional,
Tuple,
Type,
Union,
cast,
)
@ -1281,22 +1280,29 @@ class ControllableAsyncDnsServer(AsyncDnsServer):
_CONTROL_DOMAIN = "_control."
def __init__(self, commands: List[Type["ControlCommand"]]):
super().__init__()
self._control_domain = dns.name.from_text(self._CONTROL_DOMAIN)
self._commands: Dict[dns.name.Name, "ControlCommand"] = {}
for command_class in commands:
command = command_class()
command_subdomain = dns.name.Name([command.control_subdomain])
control_subdomain = command_subdomain.concatenate(self._control_domain)
try:
existing_command = self._commands[control_subdomain]
except KeyError:
self._commands[control_subdomain] = command
else:
raise RuntimeError(
f"{control_subdomain} already handled by {existing_command}"
)
@functools.cached_property
def _control_domain(self) -> dns.name.Name:
return dns.name.from_text(self._CONTROL_DOMAIN)
@functools.cached_property
def _commands(self) -> Dict[dns.name.Name, "ControlCommand"]:
return {}
def install_control_commands(self, commands: List["ControlCommand"]) -> None:
for command in commands:
self.install_control_command(command)
def install_control_command(self, command: "ControlCommand") -> None:
command_subdomain = dns.name.Name([command.control_subdomain])
control_subdomain = command_subdomain.concatenate(self._control_domain)
try:
existing_command = self._commands[control_subdomain]
except KeyError:
self._commands[control_subdomain] = command
else:
raise RuntimeError(
f"{control_subdomain} already handled by {existing_command}"
)
async def _prepare_responses(
self, qctx: QueryContext

View file

@ -106,6 +106,7 @@ class AXFRServer(DomainHandler):
if __name__ == "__main__":
server = ControllableAsyncDnsServer([ToggleResponsesCommand])
server = ControllableAsyncDnsServer()
server.install_control_command(ToggleResponsesCommand())
server.install_response_handler(AXFRServer())
server.run()