style: apply ruff and django-upgrade

This commit is contained in:
ponas committed 2026-11-02 01:05:51 +01:00
1 parent 1e128e8d9b
commit be0e621b28
18 files changed
+149 -212

No files matched your search

-1
View File
@@ -1 +0,0 @@
default_app_config = "mdb.apps.MdbAppConfig"
+24 -32
View File
@@ -54,9 +54,10 @@ class InterfaceInline(admin.TabularInline):
form = InterfaceForm
def get_queryset(self, request):
return super(InterfaceInline, self).get_queryset(request).select_related()
return super().get_queryset(request).select_related()
@admin.register(Host)
class HostAdmin(admin.ModelAdmin):
ordering = ("hostname",)
inlines = [InterfaceInline]
@@ -114,28 +115,26 @@ class HostAdmin(admin.ModelAdmin):
)
return ", ".join(addresses)
@admin.display(description="in domains")
def in_domain(self, host):
domains = host.interface_set.values_list("domain__domain_name", flat=True)
return ",".join(domains)
in_domain.short_description = "in domains"
@admin.display(boolean=True)
def ipv6_enabled(self, host):
ipv6_ifs = host.interface_set.annotate(num_ipv6=Count("ip6address")).filter(
num_ipv6__gt=0
)
return ipv6_ifs.exists()
ipv6_enabled.boolean = True
# FIXME: custom filter to sort custom field
@staticmethod
def _get_host_warning_message(host, ifs):
ifs_str = ", ".join([_if.name for _if in ifs])
return "Host {} might not be installable since interface(s) {} has no IP set.".format(
host, ifs_str
)
return f"Host {host} might not be installable since interface(s) {ifs_str} has no IP set."
@admin.action(description="Mark selected hosts as PXE installable")
def set_installable(self, request, queryset):
invalid_ifs_per_host = []
@@ -153,10 +152,8 @@ class HostAdmin(admin.ModelAdmin):
request, message="\n".join(invalid_ifs_per_host), level=messages.WARNING
)
set_installable.short_description = "Mark selected hosts as PXE installable"
def get_queryset(self, request):
return super(HostAdmin, self).get_queryset(request).select_related()
return super().get_queryset(request).select_related()
class Ip4AddressInline(admin.TabularInline):
@@ -164,11 +161,10 @@ class Ip4AddressInline(admin.TabularInline):
extra = 0
readonly_fields = ["address", "assigned_to_host"]
@admin.display(description="Assigned to Host")
def assigned_to_host(self, addr):
return addr.interface.host
assigned_to_host.short_description = "Assigned to Host"
class DhcpOptionInline(admin.TabularInline):
model = DhcpOption
@@ -180,6 +176,7 @@ class DhcpCustomFieldInline(admin.TabularInline):
extra = 0
@admin.register(Ip4Subnet)
class Ip4SubnetAdmin(SerialFieldAdminMixin, admin.ModelAdmin):
list_display = [
"name",
@@ -194,32 +191,32 @@ class Ip4SubnetAdmin(SerialFieldAdminMixin, admin.ModelAdmin):
readonly_fields = ["domain_name"]
inlines = [DhcpOptionInline, DhcpCustomFieldInline, Ip4AddressInline]
@admin.display(description="# addresses")
def num_addresses(self, ip4subnet):
subnet = ipaddress.IPv4Network(ip4subnet.network + "/" + ip4subnet.netmask)
return subnet.num_addresses
@admin.display(description="broadcast")
def broadcast_address(self, ip4subnet):
subnet = ipaddress.IPv4Network(ip4subnet.network + "/" + ip4subnet.netmask)
return subnet.broadcast_address
@admin.display(description="first address")
def first_address(self, ip4subnet):
subnet = ipaddress.IPv4Network(ip4subnet.network + "/" + ip4subnet.netmask)
return next(subnet.hosts())
@admin.display(description="last address")
def last_address(self, ip4subnet):
subnet = ipaddress.IPv4Network(ip4subnet.network + "/" + ip4subnet.netmask)
return list(subnet.hosts())[-1]
broadcast_address.short_description = "broadcast"
num_addresses.short_description = "# addresses"
first_address.short_description = "first address"
last_address.short_description = "last address"
def get_queryset(self, request):
# TODO: optimize
return super(Ip4SubnetAdmin, self).get_queryset(request)
return super().get_queryset(request)
@admin.register(Ip6Subnet)
class Ip6SubnetAdmin(SerialFieldAdminMixin, admin.ModelAdmin):
list_display = [
"name",
@@ -234,23 +231,22 @@ class Ip6SubnetAdmin(SerialFieldAdminMixin, admin.ModelAdmin):
def _network(self, ip6subnet):
return ipaddress.IPv6Network(
"%s::/%d" % (ip6subnet.network, ip6subnet.netmask), strict=False
f"{ip6subnet.network}::/{ip6subnet.netmask:d}", strict=False
)
@admin.display(description="# addresses")
def num_addresses(self, ip6subnet):
return self._network(ip6subnet).num_addresses
@admin.display(description="first address")
def first_address(self, ip6subnet):
return self._network(ip6subnet).network_address
@admin.display(description="last address")
def last_address(self, ip6subnet):
# subnet[-1] is O(1); never materialize hosts() for a /64.
return self._network(ip6subnet)[-1]
num_addresses.short_description = "# addresses"
first_address.short_description = "first address"
last_address.short_description = "last address"
class DomainSrvRecordInline(admin.TabularInline):
model = DomainSrvRecord
@@ -277,6 +273,7 @@ class DomainAAAARecordInline(admin.TabularInline):
extra = 0
@admin.register(Domain)
class DomainAdmin(SerialFieldAdminMixin, admin.ModelAdmin):
inlines = [
DomainSrvRecordInline,
@@ -298,6 +295,7 @@ class DomainAdmin(SerialFieldAdminMixin, admin.ModelAdmin):
models.IntegerField: {"widget": NumberInput(attrs={"style": "width: 10em"})},
}
@admin.display(description="Num Records")
def num_records(self, domain):
host_a_records = Host.objects.filter(interface__domain=domain).count()
size = {
@@ -306,15 +304,15 @@ class DomainAdmin(SerialFieldAdminMixin, admin.ModelAdmin):
"TXT": domain.domaintxtrecord_set.count(),
"A": domain.domainarecord_set.count() + host_a_records,
}
return ", ".join(["{}: {}".format(k, v) for k, v in size.items()])
num_records.short_description = "Num Records"
return ", ".join([f"{k}: {v}" for k, v in size.items()])
@admin.register(HostType)
class HostTypeAdmin(admin.ModelAdmin):
list_display = ["host_type", "description", "num_members"]
@admin.register(OperatingSystem)
class OperatingSystemAdmin(admin.ModelAdmin):
list_display = [
"name",
@@ -326,14 +324,8 @@ class OperatingSystemAdmin(admin.ModelAdmin):
]
admin.site.register(Domain, DomainAdmin)
admin.site.register(Host, HostAdmin)
admin.site.register(Ip4Subnet, Ip4SubnetAdmin)
admin.site.register(Ip6Subnet, Ip6SubnetAdmin)
admin.site.register(Ip6Address)
admin.site.register(Nameserver)
admin.site.register(MailExchange)
admin.site.register(OperatingSystem, OperatingSystemAdmin)
admin.site.register(HostType, HostTypeAdmin)
admin.site.register(DhcpConfig)
admin.site.register(DhcpOption)
+4 -6
View File
@@ -22,7 +22,7 @@ class SerialNumberInput(NumberInput):
def render(self, name, value, attrs=None, renderer=None):
field = super().render(name, value, attrs, renderer)
field_id = (attrs or {}).get("id", "id_%s" % name)
field_id = (attrs or {}).get("id", f"id_{name}")
extra = """
<button type="button" class="button" style="margin-left: 0.5em;"
onclick="mdbBumpSerial('__ID__')">Today + next serial</button>
@@ -44,9 +44,7 @@ if (typeof window.mdbBumpSerial === 'undefined') {
};
}
</script>
""".replace(
"__ID__", field_id
)
""".replace("__ID__", field_id)
return mark_safe(field + extra)
@@ -55,7 +53,7 @@ class Ip4ModelChoiceField(ModelChoiceField):
if not hasattr(obj, "interface"):
return obj.address
return "{} ({})".format(obj.address, obj.interface.host.hostname)
return f"{obj.address} ({obj.interface.host.hostname})"
class InterfaceForm(forms.ModelForm):
@@ -71,7 +69,7 @@ class InterfaceForm(forms.ModelForm):
raise ValidationError("Macaddr already in this subnet")
def __init__(self, *args, **kwargs):
super(InterfaceForm, self).__init__(*args, **kwargs)
super().__init__(*args, **kwargs)
# Join Ip4Address with tables interface and host,
# before printing the select labels that walk those two relations.
+5 -6
View File
@@ -42,7 +42,7 @@ class Command(BaseCommand):
config = service.get_config()
if not service.needs_update(config):
self.stdout.write("Nothing to do, serial is the same: %s\n" % config.serial)
self.stdout.write(f"Nothing to do, serial is the same: {config.serial}\n")
return
if output is None:
@@ -64,20 +64,19 @@ class Command(BaseCommand):
self.stdout.write("fail\n")
message = (
"Failed to restart isc-dhcp-server, please inspect...\n\n"
"%s@%s$ %s\n%s"
% (os.getlogin(), os.uname()[1], service.restart_command, output)
f"{os.getlogin()}@{os.uname()[1]}$ {service.restart_command}\n{output}"
)
discord.notify(
subject="⚠️ FAIL: dhcpd %s -> %s" % (old_serial, config.active_serial),
subject=f"⚠️ FAIL: dhcpd {old_serial} -> {config.active_serial}",
message=message,
preformatted=diff or None,
)
raise SystemExit(1)
self.stdout.write("ok\n")
message = "Updated dhcpd configuration to serial %d" % config.active_serial
message = f"Updated dhcpd configuration to serial {config.active_serial:d}"
discord.notify(
subject="✅ Success: dhcpd %s -> %s" % (old_serial, config.active_serial),
subject=f"✅ Success: dhcpd {old_serial} -> {config.active_serial}",
message=message,
preformatted=diff or None,
)
+3 -10
View File
@@ -48,7 +48,7 @@ class Command(BaseCommand):
if zone_name:
zone = service.find_zone(zone_name)
if zone is None:
self.stderr.write("error: zone '%s' not found\n" % zone_name)
self.stderr.write(f"error: zone '{zone_name}' not found\n")
raise SystemExit(1)
zones = [zone]
else:
@@ -107,7 +107,7 @@ class Command(BaseCommand):
self.stdout.write("ok\n")
for zone, info in changes.items():
message = "Successfully updated bind zone file for `%s`" % zone
message = f"Successfully updated bind zone file for `{zone}`"
if info["errors"]:
message += "\n\n" + info["errors"]
discord.notify(
@@ -120,14 +120,7 @@ class Command(BaseCommand):
self, summary, failed_command, output, zones, errors="", diffs=""
):
zone_list = "\n".join("- " + z for z in zones)
message = "%s, please inspect...\n\n%s@%s$ %s\n%s\n%s" % (
summary,
os.getlogin(),
os.uname()[1],
failed_command,
output,
zone_list,
)
message = f"{summary}, please inspect...\n\n{os.getlogin()}@{os.uname()[1]}$ {failed_command}\n{output}\n{zone_list}"
if errors:
message += "\n\n" + errors
discord.notify(
+75 -102
View File
@@ -54,33 +54,33 @@ class Domain(models.Model):
return self.domain_name == other.domain_name
def zone_file_contents(self):
content = "; serial:%d\n" % self.domain_serial
content += "; zone file for %s\n" % self.domain_name
content += "; %s\n" % datetime.datetime.now()
content += "; filename: %s\n" % self.domain_filename
content += "$TTL %s\n" % self.domain_ttl
content += "@ IN SOA %s. %s. (\n" % (
content = f"; serial:{self.domain_serial:d}\n"
content += f"; zone file for {self.domain_name}\n"
content += f"; {datetime.datetime.now()}\n"
content += f"; filename: {self.domain_filename}\n"
content += f"$TTL {self.domain_ttl}\n"
content += "@ IN SOA {}. {}. (\n".format(
self.domain_soa,
self.domain_admin.replace("@", "."),
)
content += "\t%d\t; serial\n" % self.domain_serial
content += "\t%d\t; refresh\n" % self.domain_refresh
content += "\t%d\t; retry\n" % self.domain_retry
content += "\t%d\t; expire\n" % self.domain_expire
content += "\t%d )\t; minimum ttl\n" % self.domain_minimum_ttl
content += f"\t{self.domain_serial:d}\t; serial\n"
content += f"\t{self.domain_refresh:d}\t; refresh\n"
content += f"\t{self.domain_retry:d}\t; retry\n"
content += f"\t{self.domain_expire:d}\t; expire\n"
content += f"\t{self.domain_minimum_ttl:d} )\t; minimum ttl\n"
content += ";\n"
for nameserver in self.domain_nameservers.all():
content += "@\tIN\tNS\t%s.\n" % nameserver.hostname
content += f"@\tIN\tNS\t{nameserver.hostname}.\n"
for mx in self.domain_mailexchanges.all():
content += "@\t\tMX\t%d %s.\n" % (mx.priority, mx.hostname)
content += f"@\t\tMX\t{mx.priority:d} {mx.hostname}.\n"
if self.domain_ipaddr is not None:
content += "@\tIN\tA\t%s\n" % self.domain_ipaddr
content += f"@\tIN\tA\t{self.domain_ipaddr}\n"
if self.domain_ip6addr:
content += "@\tIN\tAAAA\t%s\n" % self.domain_ip6addr
content += f"@\tIN\tAAAA\t{self.domain_ip6addr}\n"
content += "; SRV records\n"
@@ -100,7 +100,7 @@ class Domain(models.Model):
content += "; CNAME records \n"
for cname in self.domaincnamerecord_set.all():
content += "%s\tIN\tCNAME\t%s\n" % (cname.name, cname.target)
content += f"{cname.name}\tIN\tCNAME\t{cname.target}\n"
content += "; TXT records \n"
@@ -112,16 +112,12 @@ class Domain(models.Model):
for interface in self.interface_set.all():
host = interface.host
if interface.ip4address:
content += "%-20s\tIN\tA\t%s\n" % (
host.hostname,
interface.ip4address.address,
content += (
f"{host.hostname:<20}\tIN\tA\t{interface.ip4address.address}\n"
)
for ipv6addr in interface.ip6address_set.all():
content += "%-20s\tIN\tAAAA\t%s\n" % (
host.hostname,
ipv6addr.full_address(),
)
content += f"{host.hostname:<20}\tIN\tAAAA\t{ipv6addr.full_address()}\n"
if host.allow_email and interface.ip4address:
content += f'{host.hostname}\tTXT\t"v=spf1 mx ip4:{interface.ip4address.address} -all"\n'
@@ -143,15 +139,7 @@ class DomainSrvRecord(models.Model):
created_date = models.DateTimeField(auto_now_add=True)
def as_record(self):
return "{}.{}.{}. IN SRV {} {} {} {}.".format(
self.srvce,
self.prot,
self.name,
self.priority,
self.weight,
self.port,
self.target,
)
return f"{self.srvce}.{self.prot}.{self.name}. IN SRV {self.priority} {self.weight} {self.port} {self.target}."
def __str__(self):
return self.as_record()
@@ -167,7 +155,7 @@ class DomainTxtRecord(models.Model):
created_date = models.DateTimeField(auto_now_add=True)
def as_record(self):
return "{} TXT {}".format(self.name, self.target)
return f"{self.name} TXT {self.target}"
def __str__(self):
return self.as_record()
@@ -180,7 +168,7 @@ class DomainCnameRecord(models.Model):
created_date = models.DateTimeField(auto_now_add=True)
def as_record(self):
return "{} IN CNAME {}".format(self.name, self.target)
return f"{self.name} IN CNAME {self.target}"
def __str__(self):
return self.as_record()
@@ -193,7 +181,7 @@ class DomainARecord(models.Model):
created_date = models.DateTimeField(auto_now_add=True)
def as_record(self):
return "{} IN A {}".format(self.name, self.target)
return f"{self.name} IN A {self.target}"
def __str__(self):
return self.as_record()
@@ -206,7 +194,7 @@ class DomainAAAARecord(models.Model):
created_date = models.DateTimeField(auto_now_add=True)
def as_record(self):
return "{} IN AAAA {}".format(self.name, self.target)
return f"{self.name} IN AAAA {self.target}"
def __str__(self):
return self.as_record()
@@ -223,29 +211,28 @@ class DhcpConfig(models.Model):
log_facility = models.CharField(max_length=255)
def dhcpd_configuration(self):
content = "# Autogenerated configuration %s\n" % datetime.datetime.now()
content = f"# Autogenerated configuration {datetime.datetime.now()}\n"
if self.authoritative:
content += "authoritative;\n"
content += "default-lease-time %d;\n" % self.default_lease_time
content += "max-lease-time %d;\n" % self.max_lease_time
content += "log-facility %s;\n" % self.log_facility
content += "ddns-update-style %s;\n" % self.ddns_update_style
content += f"default-lease-time {self.default_lease_time:d};\n"
content += f"max-lease-time {self.max_lease_time:d};\n"
content += f"log-facility {self.log_facility};\n"
content += f"ddns-update-style {self.ddns_update_style};\n"
# time to write the subnet definitions
for subnet in self.ip4subnet_set.all():
content += "\n# %s\n" % subnet.name
content += "subnet %s netmask %s {\n" % (subnet.network, subnet.netmask)
content += f"\n# {subnet.name}\n"
content += f"subnet {subnet.network} netmask {subnet.netmask} {{\n"
for option in subnet.dhcpoption_set.all():
content += "\toption %s %s;\n" % (option.key, option.value)
content += f"\toption {option.key} {option.value};\n"
for option in subnet.dhcpcustomfield_set.all():
content += "\t%s;\n" % option.value
content += f"\t{option.value};\n"
if subnet.dhcp_dynamic:
content += "\trange %s %s;\n" % (
subnet.dhcp_dynamic_start,
subnet.dhcp_dynamic_end,
content += (
f"\trange {subnet.dhcp_dynamic_start} {subnet.dhcp_dynamic_end};\n"
)
content += "}\n"
@@ -259,14 +246,13 @@ class DhcpConfig(models.Model):
):
continue
_if = ip4address.interface
content += "\nhost %s {\n" % _if.host.hostname
content += "\thardware ethernet %s;\n" % _if.macaddr
content += "\tfixed-address %s.%s;\n" % (
_if.host.hostname,
_if.domain.domain_name,
content += f"\nhost {_if.host.hostname} {{\n"
content += f"\thardware ethernet {_if.macaddr};\n"
content += (
f"\tfixed-address {_if.host.hostname}.{_if.domain.domain_name};\n"
)
if len(_if.pxe_filename) > 0:
content += '\tfilename "%s";\n' % _if.pxe_filename
content += f'\tfilename "{_if.pxe_filename}";\n'
content += "}\n"
return content
@@ -307,30 +293,30 @@ class Ip6Subnet(models.Model):
Delegation in ip6.arpa happens on nibble (4-bit) boundaries, so the
prefix is truncated down to whole nibbles.
"""
network = ipaddress.IPv6Address("%s::" % self.network)
network = ipaddress.IPv6Address(f"{self.network}::")
nibbles = self.netmask // 4
return ".".join(network.exploded.replace(":", "")[:nibbles])[::-1] + ".ip6.arpa"
def zone_file_contents(self, generate_unassigned=False):
content = ""
content += "; zone file for %s\n" % self.domain_name
content += "; %s\n" % datetime.datetime.now()
content += "; filename: %s\n" % self.domain_filename
content += "$TTL %s\n" % self.domain_ttl
content += "@ IN SOA %s. %s. (\n" % (
content += f"; zone file for {self.domain_name}\n"
content += f"; {datetime.datetime.now()}\n"
content += f"; filename: {self.domain_filename}\n"
content += f"$TTL {self.domain_ttl}\n"
content += "@ IN SOA {}. {}. (\n".format(
self.domain_soa,
self.domain_admin.replace("@", "."),
)
content += "\t%d\t; serial\n" % self.domain_serial
content += "\t%d\t; refresh\n" % self.domain_refresh
content += "\t%d\t; retry\n" % self.domain_retry
content += "\t%d\t; expire\n" % self.domain_expire
content += "\t%d )\t; minimum ttl\n" % self.domain_minimum_ttl
content += f"\t{self.domain_serial:d}\t; serial\n"
content += f"\t{self.domain_refresh:d}\t; refresh\n"
content += f"\t{self.domain_retry:d}\t; retry\n"
content += f"\t{self.domain_expire:d}\t; expire\n"
content += f"\t{self.domain_minimum_ttl:d} )\t; minimum ttl\n"
content += ";\n"
content += ";\n"
for nameserver in self.domain_nameservers.all():
content += "@\tIN\tNS\t%s.\n" % nameserver.hostname
content += f"@\tIN\tNS\t{nameserver.hostname}.\n"
content += ";\n"
@@ -343,14 +329,13 @@ class Ip6Subnet(models.Model):
if addr.interface.domain is None:
continue
hostname = "%s.%s" % (
addr.interface.host.hostname,
addr.interface.domain.domain_name,
hostname = (
f"{addr.interface.host.hostname}.{addr.interface.domain.domain_name}"
)
ip = ipaddress.IPv6Address(self.network + addr.address)
ip = ".".join(ip.exploded.replace(":", "")[16:])[::-1]
content += "%s\tPTR\t%s.\n" % (ip, hostname)
content += f"{ip}\tPTR\t{hostname}.\n"
# addr = ipaddr.IPv6Address("%s%s" % (interface.
# content += "%20s\tIN\tPTR\t%s.\n" % (addr.address.split(".")[3], hostname)
@@ -401,51 +386,43 @@ class Ip4Subnet(models.Model):
This is the apex BIND expects -- it must match the zone name in
named.conf or rndc refuses to load the file.
"""
net = ipaddress.IPv4Network(
"%s/%s" % (self.network, self.netmask), strict=False
)
net = ipaddress.IPv4Network(f"{self.network}/{self.netmask}", strict=False)
o1, o2, o3, o4 = str(net.network_address).split(".")
rev = "%s.%s.%s" % (o3, o2, o1)
rev = f"{o3}.{o2}.{o1}"
if net.prefixlen > 24:
return "%s-%d.%s.in-addr.arpa" % (o4, net.prefixlen, rev)
return "%s.in-addr.arpa" % rev
return f"{o4}-{net.prefixlen:d}.{rev}.in-addr.arpa"
return f"{rev}.in-addr.arpa"
def zone_file_contents(self, generate_unassigned=False):
content = ""
content += "; zone file for %s\n" % self.domain_name
content += "; %s\n" % datetime.datetime.now()
content += "; filename: %s\n" % self.domain_filename
content += "$TTL %s\n" % self.domain_ttl
content += "@ IN SOA %s. %s. (\n" % (
content += f"; zone file for {self.domain_name}\n"
content += f"; {datetime.datetime.now()}\n"
content += f"; filename: {self.domain_filename}\n"
content += f"$TTL {self.domain_ttl}\n"
content += "@ IN SOA {}. {}. (\n".format(
self.domain_soa,
self.domain_admin.replace("@", "."),
)
content += "\t%d\t; serial\n" % self.domain_serial
content += "\t%d\t; refresh\n" % self.domain_refresh
content += "\t%d\t; retry\n" % self.domain_retry
content += "\t%d\t; expire\n" % self.domain_expire
content += "\t%d )\t; minimum ttl\n" % self.domain_minimum_ttl
content += f"\t{self.domain_serial:d}\t; serial\n"
content += f"\t{self.domain_refresh:d}\t; refresh\n"
content += f"\t{self.domain_retry:d}\t; retry\n"
content += f"\t{self.domain_expire:d}\t; expire\n"
content += f"\t{self.domain_minimum_ttl:d} )\t; minimum ttl\n"
content += ";\n"
content += ";\n"
for nameserver in self.domain_nameservers.all():
content += "@\tIN\tNS\t%s.\n" % nameserver.hostname
content += f"@\tIN\tNS\t{nameserver.hostname}.\n"
content += ";\n"
for addr in self.ip4address_set.all():
""" Get PTR records for each v4 address in subnet """
if hasattr(addr, "interface") and addr.interface.domain:
hostname = "%s.%s" % (
addr.interface.host.hostname,
addr.interface.domain.domain_name,
)
content += "%-20s\tIN\tPTR\t%s.\n" % (
addr.address.split(".")[3],
hostname,
)
hostname = f"{addr.interface.host.hostname}.{addr.interface.domain.domain_name}"
content += f"{addr.address.split('.')[3]:<20}\tIN\tPTR\t{hostname}.\n"
elif not hasattr(addr, "interface") and generate_unassigned:
content += "%s\tIN\tPTR\t%s.%s\n" % (
content += "{}\tIN\tPTR\t{}.{}\n".format(
addr.address,
addr.address.split(".")[3],
"dhcp.neuf.no.",
@@ -580,7 +557,7 @@ class Interface(models.Model):
domain = models.ForeignKey(Domain, models.CASCADE)
def __str__(self):
return "%s (%s on %s)" % (self.macaddr, self.name, self.host.hostname)
return f"{self.macaddr} ({self.name} on {self.host.hostname})"
def ipv6_enabled(self):
return self.ip6address_set.exists()
@@ -595,11 +572,7 @@ class Ip6Address(models.Model):
return self.subnet.network + self.address
def __str__(self):
return "%s (%s on %s)" % (
self.full_address(),
self.interface.name,
self.interface.host.hostname,
)
return f"{self.full_address()} ({self.interface.name} on {self.interface.host.hostname})"
class Meta:
verbose_name = "IPv6 address"
+3 -4
View File
@@ -1,6 +1,7 @@
import ipaddress
from django.conf import settings
from django.template.loader import render_to_string
import ipaddress
def render_pxelinux_cfg(_if, host):
@@ -35,9 +36,7 @@ def get_pxe_filename(interface, filename_format="mac_addr"):
Mac or IP address based PXE filename
Ref: http://www.syslinux.org/wiki/index.php/PXELINUX#Examples
"""
ip_addr_in_hex = "{:02X}".format(
int(ipaddress.IPv4Address(interface.ip4address.address))
)
ip_addr_in_hex = f"{int(ipaddress.IPv4Address(interface.ip4address.address)):02X}"
formats = {
"ip_addr": ip_addr_in_hex,
"mac_addr": "01-{}".format(interface.macaddr.lower().replace(":", "-")),
+2 -2
View File
@@ -33,9 +33,9 @@ class DhcpSyncService:
Returns (old_serial, diff).
"""
try:
with open(output_path, "r") as f:
with open(output_path) as f:
old = f.read()
except IOError:
except OSError:
old = ""
new = config.dhcpd_configuration()
+10 -21
View File
@@ -12,9 +12,9 @@ from mdb.models import Domain, Ip4Subnet, Ip6Subnet
class ZoneSyncService:
rndc_bin = shutil.which("rndc") or "/usr/sbin/rndc"
freeze_command = "%s freeze" % rndc_bin
reload_command = "%s reload" % rndc_bin
thaw_command = "%s thaw" % rndc_bin
freeze_command = f"{rndc_bin} freeze"
reload_command = f"{rndc_bin} reload"
thaw_command = f"{rndc_bin} thaw"
checkzone_bin = shutil.which("named-checkzone") or "/usr/sbin/named-checkzone"
compilezone_bin = shutil.which("named-compilezone") or "/usr/sbin/named-compilezone"
checkzone_dir = "/tmp/mdb-checkzone"
@@ -28,7 +28,7 @@ class ZoneSyncService:
def validate_binaries_exist(self):
for f in (self.checkzone_bin, self.compilezone_bin, self.rndc_bin):
if not os.path.isfile(f) and not self.debug:
print("ERROR: no such file %s, exiting..." % f)
print(f"ERROR: no such file {f}, exiting...")
sys.exit(1)
def run_commands(self, commands):
@@ -76,9 +76,9 @@ class ZoneSyncService:
zonetype = getattr(zone._meta, "verbose_name", None) or zone._meta.model_name
try:
with open(zone.domain_filename, "r") as f:
with open(zone.domain_filename) as f:
old = f.read()
except IOError:
except OSError:
old = ""
# Dynamic updates (ACME dns-01 challenges) raise the serial in the
@@ -97,13 +97,8 @@ class ZoneSyncService:
}
print(
"updating %s %s [%d -> %d]"
% (
zonetype,
zone.domain_name,
zone.domain_active_serial,
zone.domain_serial,
)
f"updating {zonetype} {zone.domain_name} "
f"[{zone.domain_active_serial:d} -> {zone.domain_serial:d}]"
)
# Normalize to BIND's canonical format so the on-disk file and our
@@ -161,11 +156,7 @@ class ZoneSyncService:
tmpfile = f.name
try:
cmd = "%s -f text -F text -s relative -i none -o - %s %s" % (
self.compilezone_bin,
zone.domain_name,
tmpfile,
)
cmd = f"{self.compilezone_bin} -f text -F text -s relative -i none -o - {zone.domain_name} {tmpfile}"
status, output = getstatusoutput(cmd + " 2>/dev/null")
if status == 0:
return output
@@ -189,6 +180,4 @@ class ZoneSyncService:
with open(zonefile, "w") as f:
f.write(contents)
return getstatusoutput(
"%s %s %s" % (self.checkzone_bin, zone.domain_name, zonefile)
)
return getstatusoutput(f"{self.checkzone_bin} {zone.domain_name} {zonefile}")
+8 -7
View File
@@ -1,12 +1,13 @@
import ipaddress
import logging
import os
import uuid
from django.conf import settings
from django.db.models.signals import pre_delete, post_save, pre_save
from django.dispatch import receiver
import ipaddress
from mdb.models import Interface, Host, Ip6Subnet, Ip4Subnet, Ip4Address
from django.conf import settings
from django.db.models.signals import post_save, pre_delete, pre_save
from django.dispatch import receiver
from mdb.models import Host, Interface, Ip4Address, Ip4Subnet, Ip6Subnet
from mdb.utils import format_domain_serial_and_add_one
logger = logging.getLogger(__name__)
@@ -106,7 +107,7 @@ def create_pxe_key_and_write_pxe_files_when_host_changes(
# Write pxeconfig file to disk
with open(path, "w+") as f:
f.write(pxe_file)
logger.info("Created or updated {}".format(path))
logger.info(f"Created or updated {path}")
os.chmod(path, 0o644) # Not writeable for all
# Also set necessary fields for pxe installation
@@ -128,7 +129,7 @@ def create_pxe_key_and_write_pxe_files_when_host_changes(
if os.path.exists(path):
# Delete pxeconfig file from disk
os.unlink(path)
logger.info("deleted {}".format(path))
logger.info(f"deleted {path}")
@receiver(pre_delete, sender=Interface)
+1 -1
View File
@@ -4,4 +4,4 @@ timeout 0
label ninja
kernel {{kernel}}
append ramdisk_size=14984 locale=en_US console-setup/ask_detect=false keyboard-configuration/layoutcode=no netcfg/wireless_wep= netcfg_choose_interface=eth0 netcfg/get_hostname= preseed/url={{preseed_config_url}} vga=normal snop={{host.pxe_key}} initrd={{initrd}} -- quiet
append ramdisk_size=14984 locale=en_US console-setup/ask_detect=false keyboard-configuration/layoutcode=no netcfg/wireless_wep= netcfg_choose_interface=eth0 netcfg/get_hostname= preseed/url={{preseed_config_url}} vga=normal snop={{host.pxe_key}} initrd={{initrd}} -- quiet
+4 -4
View File
@@ -5,7 +5,7 @@ import re
def format_domain_serial_and_add_one(serial):
today = datetime.datetime.now()
res = re.findall(
r"^%4d%02d%02d(\d\d)$" % (today.year, today.month, today.day),
rf"^{today.year:4d}{today.month:02d}{today.day:02d}(\d\d)$",
str(serial),
re.DOTALL,
)
@@ -14,7 +14,7 @@ def format_domain_serial_and_add_one(serial):
""" This probably means that the serial is malformed
or the date is wrong. We assume that if the date is wrong,
it is in the past. Just create a new serial starting from 1."""
return "%4d%02d%02d%02d" % (today.year, today.month, today.day, 1)
return f"{today.year:4d}{today.month:02d}{today.day:02d}01"
elif len(res) == 1:
""" The serial contains todays date, just update it. """
try:
@@ -28,7 +28,7 @@ def format_domain_serial_and_add_one(serial):
else:
number += 1
return "%4d%02d%02d%02d" % (today.year, today.month, today.day, number)
return f"{today.year:4d}{today.month:02d}{today.day:02d}{number:02d}"
else:
""" Just return the first serial for today. """
return "%4d%02d%02d%02d" % (today.year, today.month, today.day, 1)
return f"{today.year:4d}{today.month:02d}{today.day:02d}01"
+2 -1
View File
@@ -1,6 +1,7 @@
from django.core.validators import RegexValidator
import re
from django.core.validators import RegexValidator
# http://en.wikipedia.org/wiki/Hostname#Restrictions_on_valid_host_names
hostname_re = re.compile(r"^(?!-)[-a-z0-9]+(?<!-)$", re.IGNORECASE)
validate_hostname = RegexValidator(
+3 -2
View File
@@ -1,13 +1,14 @@
from django.contrib import admin
from django.contrib.staticfiles.urls import staticfiles_urlpatterns
from django.urls import re_path, path
from django.urls import path
from django.views.generic.base import RedirectView
from mdb.api.views import HostPXEValidate
admin.autodiscover()
urlpatterns = [
re_path(r"^$", RedirectView.as_view(url="admin/", permanent=True)),
path("", RedirectView.as_view(url="admin/", permanent=True)),
path("admin/", admin.site.urls),
path(
"api/validate-host-secret/",
+2 -1
View File
@@ -1,6 +1,7 @@
from django.core.wsgi import get_wsgi_application
import os
from django.core.wsgi import get_wsgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mdbsite.settings")
application = get_wsgi_application()
-8
View File
@@ -53,11 +53,3 @@ def db_domain(db, tmp_path):
)
DomainTxtRecord.objects.create(domain=domain, name="@", target='"v=spf1 mx ~all"')
return domain
def load_command_class():
path = pathlib.Path(__file__).parent.parent / "mdb/management/commands/zone-sync.py"
spec = importlib.util.spec_from_file_location("zone_sync_command", path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod.Command
+2 -4
View File
@@ -35,11 +35,9 @@ def test_validate_puppet_host_by_secret(api_client, user_with_token, host):
_, token = user_with_token
data = {
"pxe_key": host.pxe_key,
"certname": "{}.{}".format(
host.hostname, host.interface_set.first().domain.domain_name
),
"certname": f"{host.hostname}.{host.interface_set.first().domain.domain_name}",
}
api_client.credentials(HTTP_AUTHORIZATION="Token {}".format(token.key))
api_client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}")
res = api_client.post(reverse("api-validate-host-secret"), data, format="json")
assert res.status_code == 200, res.content
assert not Host.objects.get(pk=host.pk).pxe_installable
+1
View File
@@ -3,6 +3,7 @@ from io import StringIO
from unittest.mock import MagicMock, patch
import pytest
from mdb.management.commands.zone_sync import Command as CommandClass
from mdb.services.zone_sync import ZoneSyncService