Eu tenho rodado um tipo de sistema que faz DynDNS pra minhas máquinas que estão atrás de um serviço de DHCP como meu desktop que fica aqui em minha mesa, em casa.
O princípio é que essa máquina acessa uma certa URL a cada 5 minutos. Então a cada 5 minutos eu olho pro log do servidor web que roda nessa porta e pego os dados. Pego o IPv4 e o IPv6. Na verdade pego o que vier e vejo se é IPv4 ou IPv6. Depois olho no mapa de DNS se o valor mudou. Se mudou, altero o serial do map e mando um restart no serviço.
A lógica é simples. Mas eu fiz isso uns 15 anos atrás. Pra python2.Alguma-Coisa. Portei pra python3 no melhor estilo #XGH. O resultado? Problemas aqui e ali.
Vamos dar uma olhada no script anterior.
#! /usr/bin/python3 -u
# -*- encoding: utf-8 -*-
"""
DNS updater. It checks if hostname and IP are updated
on maps. If not, update accordingly and reload bind.
"""
# grep "\/\?update_dyndns=" /var/log/apache2/dyndns-access.log | sed "s/ HTTP.*//" | awk '{print $1, $NF}'
import re
import time
import os
import subprocess
import sys
LOG = "/var/log/apache2/dyndns-access.log"
DNSCONF = "/etc/bind/master/dyndns.truta.org"
DNSDATE = "/etc/bind/master/db.truta.org"
DNSFILES = [ DNSDATE, "/etc/bind/master/db.linux-br.org" ]
TRANSLATE = {
"goosfraba" : "helio",
"raspberrypi-masoso" : "rpi0",
"raspberrypi1" : "rpi1",
"raspberrypi2" : "rpi2",
"raspberrypi3" : "rpi3"
}
def debug(*msg):
if "DEBUG" in os.environ:
print("DEBUG:", *msg)
def get_latest_ips():
"""
Dictionary in format:
Domain/node : [ ipv4, ipv6]
"""
debug("get_latest_ips()")
NODE_IP = {}
with open(LOG) as logfile:
for line in logfile.readlines():
if not re.search("\/\?update_dyndns=", line):
continue
line = re.sub(" HTTP.*", "", line)
param = line.split()
ip = param[0]
nodename = re.sub("\/\?update_dyndns=", "", param[-1])
if not nodename in NODE_IP:
NODE_IP[nodename] = [ None, None ]
# ipv4?
if re.search("([0-9].*)\.([0-9].*)\.([0-9].*)\.([0-9].*)", ip):
NODE_IP[nodename][0] = ip
debug(f"{nodename} (ipv4): {ip}")
else:
NODE_IP[nodename][1] = ip
debug(f"{nodename} (ipv6): {ip}")
return NODE_IP
def apply_translate(DIC):
debug("apply_translate()")
# DIC = { translatedName : [ipv4, ipv6] }
debug("apply_translate(): DIC:\n", DIC)
tempDIC = {}
for nodename in DIC:
debug(f"apply_translate(): searching {nodename}")
if not nodename in TRANSLATE:
debug(f"{nodename} not in TRANSLATE table")
continue
dnsName = TRANSLATE[nodename]
if dnsName is None:
debug(f"{nodename} translates to None")
continue
nodeValue = DIC[nodename]
debug(f"apply_translate(): adding {dnsName} as {nodeValue}")
tempDIC[dnsName] = nodeValue
debug("apply_translate(): returning tempDIC:\n", tempDIC)
return tempDIC
def check_for_update(DIC):
debug("check_for_update()")
isupdated = False
buf = ""
debug("check_for_update(): current DNS entries")
with open(DNSCONF) as fh:
for line in fh.readlines():
if line.startswith(";") or line.startswith("#"):
buf += line
continue
line = line.rstrip()
param = line.split()
if len(param) != 3:
buf += f"{line}\n"
continue
subname, subtype, subip = param
print(f"DNS current entry: {subname}, Type: {subtype}, IP: {subip}")
if not subname in DIC:
debug(f"check_for_update(): {subname} is not into the DIC")
buf += f"{line}\n"
continue
ipv4, ipv6 = DIC[subname]
print(f"Found {subname} for update")
# ipv4
if subtype == "A":
print("Checking IPv4")
if ipv4 != subip and len(ipv4) > 0:
print(f" * Updating {subname} from {subip} to {ipv4}")
isupdated = True
buf += f"{subname}\t\t\t{subtype}\t{ipv4}\n"
continue
# ipv6
if subtype == "AAAA":
print("Checking IPv6")
if ipv6 != subip and len(ipv6) > 0:
print(f" * Updating {subname} from {subip} to {ipv6}")
isupdated = True
buf += f"{subname}\t\t\t{subtype}\t{ipv6}\n"
continue
buf += f"{line}\n"
if isupdated:
print(f"Updating file {DNSCONF}")
debug("check_for_update(): buf:", buf)
with open(DNSCONF, 'w') as fh:
fh.write(buf)
return isupdated
def get_dns_map_serial(dns_file: str) -> str:
with open(dns_file) as fh:
for line in fh.readlines():
if not re.search(" ; serial", line):
continue
timestamp = line.split()[0]
return timestamp
raise Exception(f"Failed to read serial from dns file: {dns_file}")
def get_serial_date_update(timestamp: str) -> list:
if len(timestamp) != 10 :
raise Exception(f"Wrong timestamp size (not 10): {timestamp}")
dateformat = timestamp[:8]
serial = timestamp[8:]
return (serial)
def update_serial_in_file(old_serial: str, new_serial: str, filename: str) -> None:
with open(filename) as fh:
buf = fh.read()
buf = re.sub(old_serial, new_serial, buf)
with open(filename, 'w') as fw:
fw.write(buf)
def update_timestamp(dryrun=False):
debug("update_timestamp()")
buf = ""
today = time.strftime("%Y%m%d", time.localtime())
timestamp = get_dns_map_serial(DNSDATE)
(last_update, last_serial) = get_serial_date_update(timestamp)
print("Timestamp:", timestamp)
print("Last update:", last_update)
print("Last serial:", last_serial)
original_timestamp = ""
if int(today) > int(last_update):
timestamp = f"{today}00"
else:
serial = int(last_serial) + 1
if serial < 100:
timestamp = "%s%02d" % (today, serial)
else:
timestamp = "%08d00" % (int(today) + 1)
print("New timestamp:", timestamp)
print("Updating file", DNSDATE)
debug(f"update_timestamp(): timestamp={timestamp}")
if dryrun is False:
update_serial_in_file(original_timestamp, timestamp, DNSDATE)
else:
print("-= dry-run mode =-")
print(f"Here file {DNSDATE} would be updated.")
print("Content:\n", buf)
def bind_restart(dryrun=False):
#cmd = "systemctl restart named.service"
## reload is enough
cmd = "systemctl reload named.service"
if dryrun is False:
subprocess.call(cmd.split())
else:
print("-= dry-run mode =-")
print(f"Here command would be called: {cmd}")
def main():
now = time.ctime(time.time())
print(f"Starting: {now}")
DNS = get_latest_ips()
DNS = apply_translate(DNS)
debug(DNS)
status = check_for_update(DNS)
if status:
if len(sys.argv) > 1 and sys.argv[-1] == "--help":
print(f"Use: {sys.argv[0]} [--help|--dry-run]")
sys.exit(0)
if len(sys.argv) > 1 and sys.argv[-1] == "--dry-run":
doDryRun = True
else:
doDryRun = False
update_timestamp(dryrun=doDryRun)
bind_restart(dryrun=doDryRun)
if __name__ == '__main__':
main()
O código está gigante e uma zona. E estava dando crash. Vamos então olhar por partes.
LOG = "/var/log/apache2/dyndns-access.log"
DNSCONF = "/etc/bind/master/dyndns.truta.org"
DNSDATE = ""
DNSFILES = [ DNSDATE, "/etc/bind/master/db.linux-br.org" ]
LOG é onde eu tenho de ler pra buscar o padrão de acesso.
DNSCONF é onde ficam as definições de "IP TIPO NOME" como "1.2.3.4 A helio" pro caso de IPv4.
DNSDATE é onde ficavam as atualizações pro domínio truta.org.
Eu percebi que dava pra fazer a mesma coisa com outros domínios, então depois
eu inseri junto o DNSFILES pra ter também o linux-br.org.
No final esses aquivos só precisam ser abertos pra alterar o serial.
O mapa de dns, pra entender o que é, é esse aqui pro truta.org:
❯ cat /etc/bind/master/db.truta.org
$TTL 180 ; # three minute
@ IN SOA ns1.truta.org. helio.loureiro.eng.br. (
2025080800 ; serial
43200 ; refresh (12 hours)
3600 ; retry (1 hour)
604800 ; expire (1 week)
86400 ; minimum (1 day)
)
@ NS ns1.truta.org.
@ NS ns2.afraid.org.
@ NS ns1.first-ns.de.
@ NS robotns2.second-ns.de.
@ NS robotns3.second-ns.com.
@ A 95.216.213.181
@ MX 5 mail.truta.org.
@ AAAA 2a01:4f9:c012:f3c4::1
@ TXT "v=spf1 a mx ip4:95.216.213.181 ip6:2a01:4f9:c012:f3c4::1 include:_spf.google.com -all"
mail._domainkey IN TXT "v=DKIM1; k=rsa; t=y; p=MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCpp/HlB1QAHjp6jHGCj9kfcZSjY3ipbGuU/enVjEeptlNgw2qTxRRJ6puPiW5DWCOaPEg2nC//wRazFgMuLZ3C89vF2TT57upngmNuCcLtbwNxo6G1JFo5GD92UbQgstYC+cjfrRlw56XfwzWVtojR4ZCB8PCQCguwhwunwRhJEwIDAQAB" ; ----- DKIM key mail for truta.org
; As e AAAAs
; CNAMES
www AAAA 2a01:4f9:c012:f3c4::1
www A 95.216.213.181
ns1 A 95.216.213.181
AAAA 2a01:4f9:c012:f3c4::1
mail A 95.216.213.181
AAAA 2a01:4f9:c012:f3c4::1
; DYNAMIC TRUTAS
$INCLUDE "/etc/bind/master/dyndns.truta.org";
A linha que precisa ser alterada é a que tem 2025080800 ; serial.
A parte que faz a mágica de ler os novos endereços IPs é $INCLUDE "/etc/bind/master/dyndns.truta.org";.
E se não está familiarizado com mapas de dns, o que tem depois do ";" é ignorado.
Então o bind lê somente o 2025080800.
O ; serial é um comentário pra humanos entenderem o que é aquilo.
O serial é um número inteiro.
Você pode usar qualquer número.
Só precisa incrementar quando faz alguma alteração no mapa pra poder notificar que houve mudança pro seus secundários.
A boa prática é usar YYYYMMDDXX onde:
- YYYY: ano com 4 dígitos
- MM: mês com 2 dígitos
- DD: dia com 2 dígitos
- XX: um número inteiro com 2 dígitos
Então se o mapa é atualizado no mesmo dia, você incrementa o XX. Claro que isso te limita à 100 alterações por dia, de 0 a 99. Mas não é algo como ficar atualizando tanto o mapa. A menos que você seja um provedor de cloud. Daí as coisas funcionam de jeito beeeeem diferente. Mas vamos ficar no feijão com arroz.
Vamos voltar ao script. Como está escrito com funções, fica mais fácil de olhar pra cada uma independentemente. Eu vou colocar fora de ordem pra ficar mais claro o que faz o quê. Ou o que deveria fazer o quê.
def debug(*msg):
if "DEBUG" in os.environ:
print("DEBUG:", *msg)
Isso é claramente uma gambiarra. XGH em sua melhor forma. Só pra eu ter um "debug" pra olhar dentro do programa.
def main():
now = time.ctime(time.time())
print(f"Starting: {now}")
DNS = get_latest_ips()
DNS = apply_translate(DNS)
debug(DNS)
status = check_for_update(DNS)
if status:
if len(sys.argv) > 1 and sys.argv[-1] == "--help":
print(f"Use: {sys.argv[0]} [--help|--dry-run]")
sys.exit(0)
if len(sys.argv) > 1 and sys.argv[-1] == "--dry-run":
doDryRun = True
else:
doDryRun = False
update_timestamp(dryrun=doDryRun)
bind_restart(dryrun=doDryRun)
if __name__ == '__main__':
main()
A função main é o que chama tudo.
Algumas coisas de tempo pra saber quando começa.
E efetivamente a primeira chamada aqui: DNS = get_latest_ips().
Soa razoável olhar pra esse get_latest_ips() e imaginar que volta os últimos ips.
Mas ip de quem?
Com certeza tem de voltar algo como um dicionário pra você saber qual hostname tem qual ip.
E esse DNS?
Tinha de ser em maiúsculo?
Definitivamente não.
Variáveis em maiúsculo a gente deixa pra constantes como LOG.
Mas seguimos...
Agora temos um DNS = apply_translate(DNS).
Esse é um filtro que usa o TRANSLATE lá em cima:
TRANSLATE = {
"goosfraba" : "helio",
"raspberrypi-masoso" : "rpi0",
"raspberrypi1" : "rpi1",
"raspberrypi2" : "rpi2",
"raspberrypi3" : "rpi3"
}
E isso é usado porque meu desktop, por exemplo, tem hostname "goosfraba". Mas eu quero que ele seja no fqdn como "helio.truta.org". Então eu uso essa "tradução" de hostname recebido pro que eu quero no dns.
status = check_for_update(DNS) eu provavelmente comparo com o que está em DNSCONF.
E se houve alteração, salvo em status.
Definitivamente um boolean.
Em seguinda essa parte aqui:
if status:
if len(sys.argv) > 1 and sys.argv[-1] == "--help":
print(f"Use: {sys.argv[0]} [--help|--dry-run]")
sys.exit(0)
if len(sys.argv) > 1 and sys.argv[-1] == "--dry-run":
doDryRun = True
else:
doDryRun = False
Esse é um outro XGH. Tudo pra ver se existe o parâmetro "--help". Ou se é passado "--dry-run". "dry-run" é como é chamado um "test pra ver se vai" sem realmente fazer nada. Eu provavelmente fiz isso porque a primeira implementação era em shell script. Mas mesmo em shell script isso seria XGH.
update_timestamp(dryrun=doDryRun) faz aquele update de serial nos mapas de dns se algo mudou.
A menos que o parâmetro de "dry-run" esteja lá.
E finalmente bind_restart(dryrun=doDryRun) reinicia o serviço via systemd ser não for um "dry-run".
Olhamos a lógica do que deveria fazer.
Agora vamos ver o que foi feito.
Vamos começar com o get_latest_ips( ).
def get_latest_ips():
"""
Dictionary in format:
Domain/node : [ ipv4, ipv6]
"""
debug("get_latest_ips()")
NODE_IP = {}
with open(LOG) as logfile:
for line in logfile.readlines():
if not re.search("\/\?update_dyndns=", line):
continue
line = re.sub(" HTTP.*", "", line)
param = line.split()
ip = param[0]
nodename = re.sub("\/\?update_dyndns=", "", param[-1])
if not nodename in NODE_IP:
NODE_IP[nodename] = [ None, None ]
# ipv4?
if re.search("([0-9].*)\.([0-9].*)\.([0-9].*)\.([0-9].*)", ip):
NODE_IP[nodename][0] = ip
debug(f"{nodename} (ipv4): {ip}")
else:
NODE_IP[nodename][1] = ip
debug(f"{nodename} (ipv6): {ip}")
return NODE_IP
Aparentemente a estrutura NODE_IP = {} vai ser um dicionário do tipo
hostname: [IPv4][IPv6].
with open(LOG) as logfile:
for line in logfile.readlines():
if not re.search("\/\?update_dyndns=", line):
continue
Aqui é ler o aquivo LOG.
Se a linha não tiver o padrão "/?update_dynds=", vai pra próxima linha.
line = re.sub(" HTTP.*", "", line)
param = line.split()
ip = param[0]
nodename = re.sub("\/\?update_dyndns=", "", param[-1])
A alinha recebida é algo assim:
83.233.219.150 - - [11/Aug/2025:08:54:57 +0000] "GET /?update_dyndns=goosfraba HTTP/1.1" 200 87144 "-" "curl/8.15.0"
Então o regex em line = re.sub(" HTTP.*", "", line) remove tudo depois de " HTTP", including essa parte junto.
A linha então fica algo assim:
83.233.219.150 - - [11/Aug/2025:08:54:57 +0000] "GET /?update_dyndns=goosfraba
A linha param = line.split() cria a variável param (parâmetros) com a linha separada por espaços simples.
ip = param[0] pega o primeiro parâmetro.
No exemplo seria "83.233.219.150".
nodename = re.sub("\/\?update_dyndns=", "", param[-1]) pega o último parâmetro, que no exemplo seria
"/?update_dyndns=goosfraba" e remove a parte "/?update_dyndns=".
Vai sobrar o hostname, "goosfraba".
if not nodename in NODE_IP:
NODE_IP[nodename] = [ None, None ]
Aqui olha se o dicionário NODE_IP tem a chave que é o nodename.
Eu comentei acima hostname, mas no código chamei de nodename.
Tenha em mente que eram a mesma coisa.
E se a chave não existe, eu crio com uma lista vazia.
if re.search("([0-9].*)\.([0-9].*)\.([0-9].*)\.([0-9].*)", ip):
NODE_IP[nodename][0] = ip
debug(f"{nodename} (ipv4): {ip}")
else:
NODE_IP[nodename][1] = ip
debug(f"{nodename} (ipv6): {ip}")
Finalmente eu comparo o ip se bate com o padrão ([0-9].*)\.([0-9].*)\.([0-9].*)\.([0-9].*) que
é basicamente qualquer número de 0 à 9 seguido de ponto.
Então se eu recebesse um IP inválido, como 999.9999999.999999999999.99999999, validaria.
Mas pro script XGH, foi suficiente.
Daí eu sei se é um IPv4.
Se não for, é IPv6.
Como vou lendo o arquivo LOG de cima pra baixo,
esses valores de hostname e ip é atualizado várias vezes.
Com os mesmos valores.
E como o servidor web atualiza também de cima pra baixo, a última linha deve ser a mais atual.
Não muito eficiente, mas dá certo.
XGH raiz.
O apply_translate( ) eu não vou comentar.
Já expliquei acima o que ele faz, que é traduzir hostname pra entrada no dns.
Não adiciona muita coisa.
Então vamos pro próximo.
O próximo é um XGH raiz.
Função grande.
É a check_for_update( ).
Vamos por partes.
def check_for_update(DIC):
debug("check_for_update()")
isupdated = False
buf = ""
debug("check_for_update(): current DNS entries")
with open(DNSCONF) as fh:
for line in fh.readlines():
if line.startswith(";") or line.startswith("#"):
buf += line
continue
line = line.rstrip()
param = line.split()
if len(param) != 3:
buf += f"{line}\n"
continue
Aqui é lido o arquivo DNSCONF, que é onde está o mapa "ip tipo nome" como "1.2.3.4 A helio".
Essa primeira parte descarta as linhas que começam ou com ";" ou com "#".
Na verdade copia a linha como está em buf.
A parte de baixo prepara a linha pra ser lida se tiver o padrão de ter ao menos 3 parâmetros quando separada por espaços.
Se não for, copie pro buffer e vá pra próxima linha.
subname, subtype, subip = param
print(f"DNS current entry: {subname}, Type: {subtype}, IP: {subip}")
if not subname in DIC:
debug(f"check_for_update(): {subname} is not into the DIC")
buf += f"{line}\n"
continue
Aqui já pega os 3 parâmetros que serão subname, subtype, subip.
O subname é o hostname.
Ou nodename.
Consistência não é meu forte.
Ou não era.
E se esse hostname não estiver no dicionário, salva a linha como ela era no buffer e vai pra próxima linha.
ipv4, ipv6 = DIC[subname]
print(f"Found {subname} for update")
# ipv4
if subtype == "A":
print("Checking IPv4")
if ipv4 != subip and len(ipv4) > 0:
print(f" * Updating {subname} from {subip} to {ipv4}")
isupdated = True
buf += f"{subname}\t\t\t{subtype}\t{ipv4}\n"
continue
# ipv6
if subtype == "AAAA":
print("Checking IPv6")
if ipv6 != subip and len(ipv6) > 0:
print(f" * Updating {subname} from {subip} to {ipv6}")
isupdated = True
buf += f"{subname}\t\t\t{subtype}\t{ipv6}\n"
continue
buf += f"{line}\n"
Aqui os endereços IPv4 e IPv6 são recuperados do dicionário em ipv4, ipv6 = DIC[subname].
Se o subtipo for "A", eu olho se o endereço IPv4 mudou.
Se for "AAAA", olho o IPv6.
Se alterar, eu altero isupdated pra verdadeiro.
if isupdated:
print(f"Updating file {DNSCONF}")
debug("check_for_update(): buf:", buf)
with open(DNSCONF, 'w') as fh:
fh.write(buf)
return isupdated
E finalmente, se o código foi alterado, escrevo em cima do arquivo antigo o conteúdo novo.
Até aqui tudo bem.
O código está meio rebuscado e com alguns pontos de XGH mas parece funcional.
Vamos então olhar a próxima parte, que é atualizar os mapas caso tenha tido alguma mudança.
Olhando pra update_timestamp( ) vamos ver que esse depende de outras funções, que ficam pra cima no código.
E aqui já entra um pequeno mal estar.
Eu gosto dos conselhos do Uncle Bob sobre clean code.
E prefiro ler um código de cima pra baixo.
Usou uma função ou método?
Coloque abaixo.
Na ordem em que foi chamado.
Mas com python com funções isso não funciona.
Você tem de colocar acima as funções que serão chamadas abaixo.
Então vamos lá.
def update_timestamp(dryrun=False):
debug("update_timestamp()")
buf = ""
today = time.strftime("%Y%m%d", time.localtime())
timestamp = get_dns_map_serial(DNSDATE)
Bem básico.
Pega o dia de hoje no formato "YYYMMDD".
Em seguida olha qual o serial do arquivo com get_dns_map_serial( ).
Ao contrário do que fiz até agora, eu vou parar de analizar essa função e seguir o código que ela chamou.
def get_dns_map_serial(dns_file: str) -> str:
with open(dns_file) as fh:
for line in fh.readlines():
if not re.search(" ; serial", line):
continue
timestamp = line.split()[0]
return timestamp
raise Exception(f"Failed to read serial from dns file: {dns_file}")
Esse trecho de código é claramente mais recente. Usando f-strings e mais type hints pra saber o que seria ali. E variáveis em minúsculas! E bem simples: abre o arquivo passado (um mapa de dns) e busca pela linha com "; serial". Quebra a linha em espaços e retorna o primeiro parâmetro encontrado. Não tem verificação se realmente voltou uma string, mas parece ser isso. E se não achar nada, lança uma exceção e para a execução toda. Um pouco demais, mas pelo menos não retorna algo errado e deixa destruir o mapa de dns.
Vamos então voltar a olhar a função chamadora, update_timestamp( ).
(last_update, last_serial) = get_serial_date_update(timestamp)
Novamente um código mais enxuto.
Ao invés de manter no corpo da função, chamar outra função auto-explicativa.
Então vamos olhar a get_serial_date_update( ).
def get_serial_date_update(timestamp: str) -> list:
if len(timestamp) != 10 :
raise Exception(f"Wrong timestamp size (not 10): {timestamp}")
dateformat = timestamp[:8]
serial = timestamp[8:]
return (serial)
Aqui ele verifica se o serial tem 10 caractéres, ou 10 dígitos com if len(timestamp) != 10.
Ou gera uma exceção e encerra a execução.
Pega os primeiro 8 caractéres e coloca em dateformat.
Em seguida pega os caractéres restantes e coloca em serial.
E retorna o serial.
Perceberam os problemas?
Já começa que a função chama-se get_serial_date_update( ).
Qual o update?
E deveria retornar uma lista.
Até volta, mas com um valor só, o serial.
A função que chamava esperava o quê?
Esperava isso aqui: (last_update, last_serial) = get_serial_date_update(timestamp).
Então recebe last_update mas não last_serial.
Achamos um 🪲.
Vamos então continuar em update_timestamp( ).
print("Timestamp:", timestamp)
print("Last update:", last_update)
print("Last serial:", last_serial)
original_timestamp = ""
if int(today) > int(last_update):
timestamp = f"{today}00"
else:
serial = int(last_serial) + 1
if serial < 100:
timestamp = "%s%02d" % (today, serial)
else:
timestamp = "%08d00" % (int(today) + 1)
print("New timestamp:", timestamp)
Aqui a data today é transformada em int (integer, inteiro) e comparada com last_update, que também é transformado em int.
Se for maior, mais atual, então inicializa com "00".
Se não for mais atual, então incrementa o número serial em 1.
E olha se for maior que 100, incializa o dia seguinte e começa com 00 de novo.
E finalmente:
if dryrun is False:
update_serial_in_file(original_timestamp, timestamp, DNSDATE)
Se não está rodando em dry-run, rodar esse update_serial_in_file( ).
Pelo nome é possível deduzer que seja trocar o serial antigo do arquivo pelo novo.
Mas vamos olhar o código.
def update_serial_in_file(old_serial: str, new_serial: str, filename: str) -> None:
with open(filename) as fh:
buf = fh.read()
buf = re.sub(old_serial, new_serial, buf)
with open(filename, 'w') as fw:
fw.write(buf)
E sim, é basicamente isso mesmo. Ler o arquivo todo, trocar a string antiga pela nova, e salvar.
E com isso chegamos na parte final do script:
bind_restart(dryrun=doDryRun)
É bem descritivo e faz exatamente o que diz: reinicia o serviço de dns via systemd.
def bind_restart(dryrun=False):
#cmd = "systemctl restart named.service"
## reload is enough
cmd = "systemctl reload named.service"
if dryrun is False:
subprocess.call(cmd.split())
else:
print("-= dry-run mode =-")
print(f"Here command would be called: {cmd}")
Se teve paciência de ler até aqui, espero que tenha gostado e aprendido que esse código estava bem ruim. E o que seria um código bom? Com certeza um que seja legível de cima pra baixo. Até certo ponto porque python não é tão flexível assim. Sem delongas, vou mostrar o código novo. Depois vou explicando.
#! /usr/bin/python3 -u
# -*- encoding: utf-8 -*-
"""
DNS updater. It checks if hostname and IP are updated
on maps. If not, update accordingly and reload bind.
"""
# grep "\/\?node=" /var/log/apache2/dyndns-access.log | sed "s/ HTTP.*//" | awk '{print $1, $NF}'
import re
import time
import os
import subprocess
import sys
import argparse
import logging
logger = logging.getLogger(__file__)
consoleOutputHandler = logging.StreamHandler()
logger.addHandler(consoleOutputHandler)
logger.setLevel(logging.INFO)
TRANSLATE = {
"goosfraba" : "helio",
"raspberrypi-masoso" : "rpi0",
"raspberrypi1" : "rpi1",
"raspberrypi2" : "rpi2",
"raspberrypi3" : "rpi3"
}
class DnsIPUpdate:
'''
A class that reads httpd logs for patterns in order to find a host IP,
update dns file accordingly if needed and restart dns service if updated.
'''
def __init__(self, args: argparse.Namespace):
self._logfile: str = args.logfile
self._bindfiles: list = args.bindfiles.split(",")
self._IPv4: dict = {}
self._IPv6: dict = {}
self._dryrun: bool = args.dryrun
self._dyndnsfile: str = args.dyndnsfile
self._fakesystemd: bool = args.fakesystemd
# for the serial on the dns files
self._year: str = time.strftime("%Y", time.localtime())
self._month: str = time.strftime("%m", time.localtime())
self._day: str = time.strftime("%d", time.localtime())
def update(self):
'run the update'
matched_lines: list = self.getEntriesFromLogs()
self._populateIPsData(matched_lines)
logger.debug(f'IPv4s: {self._IPv4}')
logger.debug(f'IPv6s: {self._IPv6}')
self._applyTranslation()
logger.debug(f'IPv4s: {self._IPv4}')
logger.debug(f'IPv6s: {self._IPv6}')
if not self._dryrun:
resp: bool = self._updateDNSFiles()
if resp is True:
self._restartService()
def getEntriesFromLogs(self) -> list:
'return the matching lines from logging file'
matches: list = []
with open(self._logfile) as fd:
for line in fd.readlines():
if not re.search(r"\/\?update_dyndns=", line):
continue
line = line.rstrip()
logger.debug(f"found line in log: {line}")
matches.append(line)
return matches
def _populateIPsData(self, matches):
'fill the IPv4 and IPv6 structures with IPs and hostnames'
for line in matches:
ip_addr, _, _, timestamp, timezone, method, uri, http_version, status_code, size, _, user_agent = line.split()
timestamp = timestamp[1:]
timezone = timezone[:-1]
method = method[1:]
http_version = http_version[:-1]
user_agent = user_agent[1:-1]
logger.debug(f"ip={ip_addr} timestamp={timestamp} timezone={timezone} method={method} uri={uri} proto={http_version} status_code={status_code} size={size} agent={user_agent}")
hostname = self._getHostName(uri)
if self._isIPv4(ip_addr):
self._IPv4[hostname] = ip_addr
else:
self._IPv6[hostname] = ip_addr
def _isIPv4(self, ip: str) -> bool:
'quickly finds out whether IPv4 or not - then IPv6'
if re.search("^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$", ip):
return True
return False
def _getHostName(self, uri: str) -> str:
'filter uri to return only the hostname'
hostname = re.sub(r",.*", "", uri)
hostname = re.sub(r".*update_dyndns=", "", hostname)
return hostname
def _applyTranslation(self) -> None:
'translate hostname according to table'
for old_name, new_name in TRANSLATE.items():
if old_name in self._IPv4:
logger.debug(f"Translating IPv4: from={old_name} to={new_name}")
self._IPv4[new_name] = self._IPv4[old_name]
del self._IPv4[old_name]
if old_name in self._IPv6:
logger.debug(f"Translating IPv6: from={old_name} to={new_name}")
self._IPv6[new_name] = self._IPv6[old_name]
del self._IPv6[old_name]
def _updateDNSFiles(self) -> bool:
'check and update DNS files if needed'
isUpdated = self._updateDynDNS()
if isUpdated is True:
self._updateSerialOnDNS()
return isUpdated
def _updateDynDNS(self) -> bool:
'check each entry into dyndns file and returns true if updated'
status = False # by default we don't update
output = []
with open(self._dyndnsfile) as fd:
buffer = fd.readlines()
for line in buffer:
line = line.rstrip() # remove new line
try:
hostname, _, _, dns_type, ip_addr = line.split("\t")
except ValueError:
output.append(f"{line}\n")
continue
match dns_type:
case "A":
# IPv4
if hostname in self._IPv4:
if self._IPv4[hostname] != ip_addr:
logger.info(f"updating IPv4 for {hostname}: old={ip_addr} new={self._IPv4[hostname]}")
line = f"{hostname}\t\t\tA\t{self._IPv4[hostname]}"
status = True
case "AAAA":
# IPv6
if hostname in self._IPv6:
if self._IPv6[hostname] != ip_addr:
logger.info(f"updating IPv6 for {hostname}: old={ip_addr} new={self._IPv6[hostname]}")
line = f"{hostname}\t\t\tAAAA\t{self._IPv6[hostname]}"
status = True
case _:
pass
output.append(f"{line}\n")
if status is True:
with open(self._dyndnsfile, 'w') as fw:
fw.write("".join(output))
return status
def _updateSerialOnDNS(self) -> None:
for filename in self._bindfiles:
buffer: list = []
logger.debug(f"updating serial on file: {filename}")
with open(filename) as fd:
for line in fd.readlines():
if not re.search("; serial", line):
buffer.append(line)
continue
line = line.rstrip()
logger.debug(f"line with serial number: {line}")
serial = self._sanitizeSerial(line)
logger.debug(f"serial={serial}")
new_serial = self._increaseSerial(serial)
logger.debug(f"new_serial={new_serial}")
line = f"\t\t\t\t{new_serial} ; serial\n"
logger.info(f"updating: filename={filename} old_serial={serial} new_serial={new_serial}")
buffer.append(line)
with open(filename, "w") as fw:
fw.write("".join(buffer))
def _sanitizeSerial(self, serial_nr: str) -> str:
'to remove the \t and spaces'
serial, _ = serial_nr.split(";")
serial = re.sub(r"\t", "", serial)
serial = re.sub(" ", "", serial)
return serial
def _increaseSerial(self, serial_nr: str):
'to identify and increase serial'
year = serial_nr[:4]
month = serial_nr[4:6]
day = serial_nr[6:8]
counter = serial_nr[8:]
today = f"{self._year}{self._month}{self._day}"
serial_date = f"{year}{month}{day}"
if today == serial_date:
new_counter = self._increaseString(counter)
return f"{today}{new_counter}"
else:
return f"{today}00"
def _increaseString(self, str_number: str) -> str:
'convert string to int, increase by one and return as string'
int_number: int = int(str_number)
int_number += 1
return f"{int_number}"
def _restartService(self) -> None:
'restart systemd service'
cmd = ["systemctl", "reload", "named.service"]
logger.info("restarting named.service")
if self._fakesystemd is True:
print(f"here service would be restarted: {cmd}")
else:
subprocess.call(cmd)
def getTimestamp():
return time.strftime("%Y%m%dT%H:%M:%S", time.localtime())
def main():
logger.info(f"starting: {getTimestamp()}")
parser = argparse.ArgumentParser(description='script to update DNS entries for dynamic clients found on httpd logs')
parser.add_argument("--logfile", required=True, help="the httpd log to be checked")
parser.add_argument("--loglevel", default="info", help="logging level for this script")
parser.add_argument("--bindfiles", required=True, help="files separated by \",\" to be updated to update serial")
parser.add_argument("--dyndnsfile", required=True, help="the dyndns map to update the IPs")
parser.add_argument("--dryrun", default=False, type=bool, help='run as dry-run or not (default=false)')
parser.add_argument("--fakesystemd", default=False, type=bool, help='run just a printout instead of systemd')
args = parser.parse_args()
if args.loglevel != "info":
logger.setLevel(args.loglevel.upper())
logger.debug(f"args: {args}")
dns = DnsIPUpdate(args)
dns.update()
logger.info(f"finished: {getTimestamp()}")
if __name__ == '__main__':
main()
A parte de baixo continua mais ou menos igual: chama uma função main( ) que faz tudo.
Mas agora a função main( ) usa argparse pra pegar os argumentos.
Então as constantes com os nomes dos arquivos sumiram e viraram agora argumentos.
Isso facilita escrever testes (o que não fiz ainda).
O que o código faz?
Faz isso aqui:
dns = DnsIPUpdate(args)
dns.update()
Cria um objeto DnsIPUpdate( ) e chama o método update( ).
Não mais funções mas métodos.
Parece mais simples de entender.
Espero.
Vamos então agora olhar a classe. Começando com sua inicialização.
def __init__(self, args: argparse.Namespace):
self._logfile: str = args.logfile
self._bindfiles: list = args.bindfiles.split(",")
self._IPv4: dict = {}
self._IPv6: dict = {}
self._dryrun: bool = args.dryrun
self._dyndnsfile: str = args.dyndnsfile
self._fakesystemd: bool = args.fakesystemd
# for the serial on the dns files
self._year: str = time.strftime("%Y", time.localtime())
self._month: str = time.strftime("%m", time.localtime())
self._day: str = time.strftime("%d", time.localtime())
As variáveis (atributos da classe) vêm do argparse. Todas com o "_" no início pra indicar que são internas. Não que isso importe muito uma vez que não é uma classe feita pra ser usada como biblioteca. E agora IPv4 e IPv6 são estruturas separadas. Do tipo dicionário. E pegamos ano, mês e dia separados.
Vamos então olhar o método chamado, o update( ):
def update(self):
'run the update'
matched_lines: list = self.getEntriesFromLogs()
self._populateIPsData(matched_lines)
logger.debug(f'IPv4s: {self._IPv4}')
logger.debug(f'IPv6s: {self._IPv6}')
self._applyTranslation()
logger.debug(f'IPv4s: {self._IPv4}')
logger.debug(f'IPv6s: {self._IPv6}')
if not self._dryrun:
resp: bool = self._updateDNSFiles()
if resp is True:
self._restartService()
Primeiramente que o método update( ) aparece logo abaixo da inicialização da classe.
Então você lê o código de cima pra baixo, o que é muito desejado.
E o código chama outro método em matched_lines: list = self.getEntriesFromLogs( ), que vai retornar uma lista.
Vamost então olhar o que faz getEntriesFromLogs( ).
def getEntriesFromLogs(self) -> list:
'return the matching lines from logging file'
matches: list = []
with open(self._logfile) as fd:
for line in fd.readlines():
if not re.search(r"\/\?update_dyndns=", line):
continue
line = line.rstrip()
logger.debug(f"found line in log: {line}")
matches.append(line)
return matches
Ele então abre o arquivo _logfile e guarda as linhas com o padrão "/?update_dyndns=".
E em formate de lista.
Depois retorna esses valores.
Não melhorou muito em termos de ler o arquivo todo e fazer o parsing de tudo.
Ficou só mais legível.
Voltando ao update( ) temos logo em seguida: self._populateIPsData(matched_lines).
Então vamos olhar o código de _populateIPsData( ).
def _populateIPsData(self, matches):
'fill the IPv4 and IPv6 structures with IPs and hostnames'
for line in matches:
ip_addr, _, _, timestamp, timezone, method, uri, http_version, status_code, size, _, user_agent = line.split()
timestamp = timestamp[1:]
timezone = timezone[:-1]
method = method[1:]
http_version = http_version[:-1]
user_agent = user_agent[1:-1]
logger.debug(f"ip={ip_addr} timestamp={timestamp} timezone={timezone} method={method} uri={uri} proto={http_version} status_code={status_code} size={size} agent={user_agent}")
hostname = self._getHostName(uri)
if self._isIPv4(ip_addr):
self._IPv4[hostname] = ip_addr
else:
self._IPv6[hostname] = ip_addr
Agora a linha é quebrada em espaços mas o dados são utilizados. O que é "_" significa que seja descartado. Então olhando a linha de exemplo novamente:
83.233.219.150 - - [11/Aug/2025:08:54:57 +0000] "GET /?update_dyndns=goosfraba HTTP/1.1" 200 200 "-" "curl/8.15.0"
Temos:
- ip_addr=83.233.219.150
- _=-
- _=-
- timestamp=[11/Aug/2025:08:54:57
- timezone=+0000]
- method="GET
- uri=/?update_dyndns=goosfraba
- http_version=HTTP/1.1"
- status_code=200
- size=200
- _="-"
- user_agent="curl/8.15.0"
O que vem a seguir é somente pra limpar esses parâmetros.
Servem pra alguma coisa?
Não.
Só pra ficar bonitinho no debug.
O que realmente importa são os valores de ip_addr e uri.
Em seguida é chamado hostname = self._getHostName(uri).
Vamos olhar o código do _getHostName( ).
def _getHostName(self, uri: str) -> str:
'filter uri to return only the hostname'
hostname = re.sub(r",.*", "", uri)
hostname = re.sub(r".*update_dyndns=", "", hostname)
return hostname
É somente o mesmo sanitizador usado anteriormente. Vai remover o "/?update_dyndns=" e deixar somente o hostname.
Voltando ao _populateIPsData( ), temos a chamada if self._isIPv4(ip_addr):
pra verificar se é um endereço IPv4 ou não.
def _isIPv4(self, ip: str) -> bool:
'quickly finds out whether IPv4 or not - then IPv6'
if re.search("^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$", ip):
return True
return False
O código agora olha se é um número válido dentro do range de IPv4.
Então após essa verificação, ou o dicionário _IPv4 ou o _IPv6 é populado. E segue o formato "hostname: IP". Mas cada um em seu dicionário agora.
Voltando ao update( ) temos:
logger.debug(f'IPv4s: {self._IPv4}')
logger.debug(f'IPv6s: {self._IPv6}')
self._applyTranslation()
logger.debug(f'IPv4s: {self._IPv4}')
logger.debug(f'IPv6s: {self._IPv6}')
Que é a mesmo tradução feita anteriormente.
def _applyTranslation(self) -> None:
'translate hostname according to table'
for old_name, new_name in TRANSLATE.items():
if old_name in self._IPv4:
logger.debug(f"Translating IPv4: from={old_name} to={new_name}")
self._IPv4[new_name] = self._IPv4[old_name]
del self._IPv4[old_name]
if old_name in self._IPv6:
logger.debug(f"Translating IPv6: from={old_name} to={new_name}")
self._IPv6[new_name] = self._IPv6[old_name]
del self._IPv6[old_name]
Seguindo em frente no update( ) temos o seguinte:
if not self._dryrun:
resp: bool = self._updateDNSFiles()
Vamos olhar então _updateDNSFiles( ) que faz o update já visto anteriormente.
def _updateDNSFiles(self) -> bool:
'check and update DNS files if needed'
isUpdated = self._updateDynDNS()
if isUpdated is True:
self._updateSerialOnDNS()
return isUpdated
Ele chama _updateDynDNS( ).
Vamos então ao código.
def _updateDynDNS(self) -> bool:
'check each entry into dyndns file and returns true if updated'
status = False # by default we don't update
output = []
with open(self._dyndnsfile) as fd:
buffer = fd.readlines()
for line in buffer:
line = line.rstrip() # remove new line
try:
hostname, _, _, dns_type, ip_addr = line.split("\t")
except ValueError:
output.append(f"{line}\n")
continue
Então ele abre o arquivo e lê linha por linha.
Mas dessa vez tenta popular a estrutura: hostname, _, _, dns_type, ip_addr = line.split("\t").
Se der errado, joga a linha no que será escrito depois.
E segue pra próxima linha.
Continuando.
match dns_type:
case "A":
# IPv4
if hostname in self._IPv4:
if self._IPv4[hostname] != ip_addr:
logger.info(f"updating IPv4 for {hostname}: old={ip_addr} new={self._IPv4[hostname]}")
line = f"{hostname}\t\t\tA\t{self._IPv4[hostname]}"
status = True
case "AAAA":
# IPv6
if hostname in self._IPv6:
if self._IPv6[hostname] != ip_addr:
logger.info(f"updating IPv6 for {hostname}: old={ip_addr} new={self._IPv6[hostname]}")
line = f"{hostname}\t\t\tAAAA\t{self._IPv6[hostname]}"
status = True
case _:
pass
Usa match ao invés de um if.
Poderia usar if?
Sim.
Mas achei que o match deixou o código mais organizado.
Em seguida verifica se o hostname existe no dicionário de IP.
Se existir, compara se o IP está diferente ou não.
Se estiver, atualiza a flag status pra verdadeiro pra avisar que houveram mudanças.
output.append(f"{line}\n")
if status is True:
with open(self._dyndnsfile, 'w') as fw:
fw.write("".join(output))
return status
A parte final é salvar o arquivo caso tenha havido alguma alteração.
Voltando para _updateDNSFiles( ), temos:
if isUpdated is True:
self._updateSerialOnDNS()
return isUpdated
Vamos então olhar _updateSerialOnDNS( ):
def _updateSerialOnDNS(self) -> None:
for filename in self._bindfiles:
buffer: list = []
logger.debug(f"updating serial on file: {filename}")
with open(filename) as fd:
for line in fd.readlines():
if not re.search("; serial", line):
buffer.append(line)
continue
line = line.rstrip()
logger.debug(f"line with serial number: {line}")
serial = self._sanitizeSerial(line)
logger.debug(f"serial={serial}")
new_serial = self._increaseSerial(serial)
logger.debug(f"new_serial={new_serial}")
line = f"\t\t\t\t{new_serial} ; serial\n"
logger.info(f"updating: filename={filename} old_serial={serial} new_serial={new_serial}")
buffer.append(line)
with open(filename, "w") as fw:
fw.write("".join(buffer))
Aqui já virou feijão com arroz.
O que temos de olhar é _sanitizeSerial( ),
depois _increaseSerial( ).
def _sanitizeSerial(self, serial_nr: str) -> str:
'to remove the \t and spaces'
serial, _ = serial_nr.split(";")
serial = re.sub(r"\t", "", serial)
serial = re.sub(" ", "", serial)
return serial
Sem surpresas, só um regex pra voltar o número serial.
def _increaseSerial(self, serial_nr: str):
'to identify and increase serial'
year = serial_nr[:4]
month = serial_nr[4:6]
day = serial_nr[6:8]
counter = serial_nr[8:]
today = f"{self._year}{self._month}{self._day}"
serial_date = f"{year}{month}{day}"
if today == serial_date:
new_counter = self._increaseString(counter)
return f"{today}{new_counter}"
else:
return f"{today}00"
Aqui o código mudou.
Ao invés de comparar com números inteiros, faz uma comparação de string direto.
E chama _increaseString( ) pra aumentar o valor da string.
def _increaseString(self, str_number: str) -> str:
'convert string to int, increase by one and return as string'
int_number: int = int(str_number)
int_number += 1
return f"{int_number}"
Aqui sim a string é convertida pra inteiro. Esse é incrementado por 1. Mas retorna como string.
Chegamos então na parte final do update( ) que chama o método _restartService().
def _restartService(self) -> None:
'restart systemd service'
cmd = ["systemctl", "reload", "named.service"]
logger.info("restarting named.service")
if self._fakesystemd is True:
print(f"here service would be restarted: {cmd}")
else:
subprocess.call(cmd)
Sem muita novidade, é o código pra reiniciar o serviço de dns via systemd.
E aqui acabamos com o código novo. O que achou? Eu achei melhor de manter. E funcionando sem bugs.
O próximo passo será escrever testes unitários pra ele ☺️.
