julioandClaude Opus 5.5
f8d2ebf705
Associar numa rede que não é a da casa fazia o serviço dar o Wi-Fi como conectado, mover o cabo para .34 e responder pelo wlan0: a placa sumia da rede. Agora: - a conexão só conta quando o gateway responde pelo wlan0; - falha com mensagem "conectei em X, mas não chego ao roteador"; - rotas do wlan0 com métrica 600 (o cabo continua preferido); - ignore_routes_with_linkdown: cabo desligado não segura rota. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
402 lines
19 KiB
Python
402 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Wi-Fi da placa com modo de configuração (estilo "Wi-Fi de hotel").
|
|
|
|
- Rede salva e encontrada: conecta (cliente) e, se pedido, usa o endereço fixo 192.168.1.35.
|
|
- Sem rede salva, ou a rede sumiu por alguns minutos: vira o ponto de acesso "Carinha-Config".
|
|
Quem conecta nele cai numa página (abre sozinha no celular) para escolher a rede e a senha.
|
|
- No modo configuração, sem ninguém conectado, tenta voltar à rede salva a cada 5 minutos
|
|
(faltou luz e a placa ligou antes do roteador: não fica presa aqui).
|
|
|
|
O chip (unisoc_wifi) não faz cliente e ponto de acesso ao mesmo tempo: o serviço alterna entre os
|
|
dois. O cabo de rede fica com um endereço fixo de reserva (192.168.1.34) para nunca perder o acesso.
|
|
Só usa a biblioteca padrão; hostapd e dnsmasq rodam como processos filhos.
|
|
"""
|
|
import hashlib
|
|
import html
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs
|
|
|
|
IFACE = 'wlan0'
|
|
CABO = 'end0'
|
|
DIR = Path('/etc/carinha-wifi')
|
|
REDE = DIR / 'rede.json' # rede salva (só root lê)
|
|
AJUSTES = DIR / 'ajustes.json' # nome/senha do ponto de acesso
|
|
RUN = Path('/run/carinha-wifi')
|
|
ESTADO = RUN / 'estado.json' # para o painel mostrar
|
|
NETWORK = Path('/run/systemd/network/30-carinha-wlan0.network')
|
|
NETPLAN_CABO = Path('/etc/netplan/10-dhcp-all-interfaces.yaml')
|
|
|
|
AP_IP = '192.168.4.1'
|
|
IP_FIXO, GATEWAY, DNS = '192.168.1.35', '192.168.1.1', '192.168.1.1'
|
|
IP_CABO_RESERVA = '192.168.1.34'
|
|
ESPERA_CONECTAR = 60 # s para conectar antes de desistir e virar ponto de acesso
|
|
QUEDA_TOLERADA = 180 # s sem a rede (já conectado) até virar ponto de acesso
|
|
RETENTAR_A_CADA = 300 # s no modo configuração, sem ninguém conectado, para tentar a rede salva
|
|
|
|
PADRAO_AJUSTES = dict(ap_nome='Carinha-Config', ap_senha='carinha123')
|
|
|
|
|
|
def log(*a):
|
|
print('carinha-wifi:', *a, flush=True)
|
|
|
|
|
|
def sh(*cmd, check=False, timeout=20):
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
|
if check and r.returncode:
|
|
raise RuntimeError(f'{cmd[0]} falhou: {r.stderr.strip()[:200]}')
|
|
return r
|
|
|
|
|
|
def ler_json(caminho, padrao):
|
|
try:
|
|
return json.loads(Path(caminho).read_text(encoding='utf-8'))
|
|
except (OSError, ValueError):
|
|
return padrao
|
|
|
|
|
|
def gravar(caminho, dados, modo=0o600):
|
|
caminho = Path(caminho)
|
|
caminho.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = caminho.with_suffix('.tmp')
|
|
fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, modo)
|
|
with os.fdopen(fd, 'w', encoding='utf-8') as f:
|
|
f.write(dados if isinstance(dados, str) else json.dumps(dados, ensure_ascii=False, indent=2))
|
|
tmp.replace(caminho)
|
|
|
|
|
|
class Wifi:
|
|
def __init__(self):
|
|
self.modo = None # 'cliente' | 'ap'
|
|
self.desde = time.time()
|
|
self.ok_em = 0.0
|
|
self.erro = ''
|
|
self.redes = [] # última busca: [{ssid, sinal, seguranca}]
|
|
self.filhos = {}
|
|
self.portal = None
|
|
self.lock = threading.RLock()
|
|
self.trocar_para = None # pedido vindo da página
|
|
self.anterior = False # rede salva antes do pedido (False = nenhum pedido em teste)
|
|
RUN.mkdir(parents=True, exist_ok=True)
|
|
|
|
# ------------------------------------------------------------ utilidades
|
|
def ajustes(self):
|
|
return dict(PADRAO_AJUSTES, **ler_json(AJUSTES, {}))
|
|
|
|
def rede(self):
|
|
r = ler_json(REDE, None)
|
|
return r if isinstance(r, dict) and r.get('ssid') else None
|
|
|
|
def conectado(self):
|
|
link = sh('iw', 'dev', IFACE, 'link').stdout
|
|
return 'Connected to' in link
|
|
|
|
def gateway_wlan(self):
|
|
m = re.search(r'default via (\S+)', sh('ip', '-4', 'route', 'show', 'dev', IFACE).stdout)
|
|
return m.group(1) if m else (GATEWAY if (self.rede() or {}).get('ip_fixo', True) else '')
|
|
|
|
def chega_na_rede(self):
|
|
"""Associar não basta: o roteador tem que responder PELO Wi-Fi. Uma rede que não é a da casa
|
|
(outro roteador, rede de visitante) associa normalmente e deixaria a placa inacessível."""
|
|
gw = self.gateway_wlan()
|
|
return bool(gw) and sh('ping', '-c', '1', '-W', '2', '-I', IFACE, gw, timeout=6).returncode == 0
|
|
|
|
def ip_wlan(self):
|
|
m = re.search(r'inet (\d+\.\d+\.\d+\.\d+)', sh('ip', '-4', 'addr', 'show', IFACE).stdout)
|
|
return m.group(1) if m else ''
|
|
|
|
def publicar(self):
|
|
r = self.rede()
|
|
gravar(ESTADO, dict(modo=self.modo, desde=self.desde, rede=r['ssid'] if r else None, ip=self.ip_wlan(),
|
|
erro=self.erro, ap=self.ajustes()['ap_nome']), 0o644)
|
|
|
|
def buscar_redes(self):
|
|
"""Busca redes (só funciona como cliente; no modo ponto de acesso usa a última busca)."""
|
|
sh('ip', 'link', 'set', IFACE, 'up')
|
|
saida = sh('iw', 'dev', IFACE, 'scan', timeout=30).stdout
|
|
if not saida:
|
|
saida = sh('iw', 'dev', IFACE, 'scan', 'ap-force', timeout=30).stdout
|
|
redes, atual = {}, {}
|
|
for linha in saida.splitlines():
|
|
linha = linha.strip()
|
|
if linha.startswith('BSS '):
|
|
atual = {}
|
|
elif linha.startswith('signal:'):
|
|
atual['sinal'] = float(linha.split()[1])
|
|
elif linha.startswith('SSID:'):
|
|
ssid = linha[5:].strip()
|
|
if ssid and '\\x00' not in ssid:
|
|
atual['ssid'] = ssid
|
|
if ssid not in redes or redes[ssid]['sinal'] < atual.get('sinal', -100):
|
|
redes[ssid] = dict(ssid=ssid, sinal=atual.get('sinal', -100), seguranca=True)
|
|
if redes:
|
|
self.redes = sorted(redes.values(), key=lambda r: -r['sinal'])[:20]
|
|
return self.redes
|
|
|
|
def parar_filhos(self):
|
|
for nome, p in list(self.filhos.items()):
|
|
if p.poll() is None:
|
|
p.terminate()
|
|
try:
|
|
p.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
p.kill()
|
|
self.filhos.clear()
|
|
sh('pkill', '-f', f'wpa_supplicant.*{RUN}')
|
|
|
|
# ------------------------------------------------------------ cliente
|
|
def modo_cliente(self, rede):
|
|
log('conectando em', rede['ssid'])
|
|
self.parar_portal()
|
|
self.parar_filhos()
|
|
sh('ip', 'addr', 'flush', 'dev', IFACE)
|
|
sh('ip', 'link', 'set', IFACE, 'down')
|
|
sh('iw', 'dev', IFACE, 'set', 'type', 'managed')
|
|
sh('ip', 'link', 'set', IFACE, 'up')
|
|
sh('iw', 'dev', IFACE, 'set', 'power_save', 'off') # economia de energia atrasa e derruba a conexão
|
|
ssid_hex = rede['ssid'].encode().hex()
|
|
if rede.get('senha'):
|
|
psk = hashlib.pbkdf2_hmac('sha1', rede['senha'].encode(), rede['ssid'].encode(), 4096, 32).hex()
|
|
seg = f'key_mgmt=WPA-PSK\n psk={psk}'
|
|
else:
|
|
seg = 'key_mgmt=NONE'
|
|
gravar(RUN / 'wpa.conf', f'ctrl_interface={RUN}/wpa\ncountry=BR\nnetwork={{\n ssid={ssid_hex}\n scan_ssid=1\n {seg}\n}}\n')
|
|
self.filhos['wpa'] = subprocess.Popen(['wpa_supplicant', '-i', IFACE, '-D', 'nl80211', '-c', str(RUN / 'wpa.conf')],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
if rede.get('ip_fixo', True):
|
|
# rotas do Wi-Fi com métrica maior: com o cabo ligado, o cabo sempre ganha
|
|
conteudo = (f'[Match]\nName={IFACE}\n\n[Network]\nDNS={DNS}\n\n[Address]\nAddress={IP_FIXO}/24\nRouteMetric=600\n'
|
|
f'\n[Route]\nGateway={GATEWAY}\nMetric=600\n')
|
|
else:
|
|
conteudo = f'[Match]\nName={IFACE}\n\n[Network]\nDHCP=ipv4\n\n[DHCPv4]\nRouteMetric=600\n'
|
|
gravar(NETWORK, conteudo, 0o644) # o systemd-networkd põe o endereço quando conectar
|
|
sh('networkctl', 'reload')
|
|
sh('networkctl', 'reconfigure', IFACE)
|
|
self.modo, self.desde, self.ok_em = 'cliente', time.time(), 0.0
|
|
self.publicar()
|
|
|
|
def liberar_ip_fixo_do_cabo(self):
|
|
"""O cabo usava 192.168.1.35: passa para 192.168.1.34 (reserva) para o Wi-Fi assumir o .35."""
|
|
try:
|
|
texto = NETPLAN_CABO.read_text(encoding='utf-8')
|
|
except OSError:
|
|
return
|
|
if f'- {IP_FIXO}/24' not in texto:
|
|
return
|
|
log('cabo passa para', IP_CABO_RESERVA, '(o Wi-Fi fica com', IP_FIXO + ')')
|
|
NETPLAN_CABO.with_suffix('.yaml.antes-wifi').write_text(texto, encoding='utf-8')
|
|
gravar(NETPLAN_CABO, texto.replace(f'- {IP_FIXO}/24', f'- {IP_CABO_RESERVA}/24'), 0o600)
|
|
sh('netplan', 'generate')
|
|
sh('networkctl', 'reload')
|
|
sh('networkctl', 'reconfigure', CABO)
|
|
|
|
# ------------------------------------------------------------ ponto de acesso
|
|
def modo_ap(self, motivo):
|
|
log('modo configuração:', motivo)
|
|
self.parar_filhos()
|
|
try:
|
|
NETWORK.unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
sh('networkctl', 'reload')
|
|
if self.modo != 'ap':
|
|
try:
|
|
self.buscar_redes() # ainda como cliente: guarda a lista para a página
|
|
except Exception as e:
|
|
log('busca falhou:', e)
|
|
aj = self.ajustes()
|
|
sh('ip', 'addr', 'flush', 'dev', IFACE)
|
|
sh('ip', 'link', 'set', IFACE, 'down')
|
|
sh('iw', 'dev', IFACE, 'set', 'type', '__ap')
|
|
sh('ip', 'link', 'set', IFACE, 'up')
|
|
sh('ip', 'addr', 'add', f'{AP_IP}/24', 'dev', IFACE)
|
|
gravar(RUN / 'hostapd.conf', '\n'.join([
|
|
f'interface={IFACE}', 'driver=nl80211', f'ssid={aj["ap_nome"]}', 'country_code=BR', 'hw_mode=g', 'channel=6',
|
|
'wmm_enabled=1', 'auth_algs=1', 'wpa=2', 'wpa_key_mgmt=WPA-PSK', 'rsn_pairwise=CCMP',
|
|
f'wpa_passphrase={aj["ap_senha"]}', '']))
|
|
self.filhos['hostapd'] = subprocess.Popen(['hostapd', str(RUN / 'hostapd.conf')],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
self.filhos['dnsmasq'] = subprocess.Popen([
|
|
'dnsmasq', '--keep-in-foreground', '--conf-file=/dev/null', '--no-resolv', '--no-hosts',
|
|
f'--interface={IFACE}', '--bind-interfaces', f'--listen-address={AP_IP}', '--except-interface=lo',
|
|
'--dhcp-range=192.168.4.10,192.168.4.60,255.255.255.0,1h', f'--dhcp-option=3,{AP_IP}',
|
|
f'--dhcp-option=6,{AP_IP}', f'--dhcp-option=114,http://{AP_IP}/', f'--address=/#/{AP_IP}',
|
|
f'--pid-file={RUN}/dnsmasq.pid'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
self.iniciar_portal()
|
|
self.modo, self.desde = 'ap', time.time()
|
|
self.publicar()
|
|
|
|
def alguem_no_ap(self):
|
|
return 'Station' in sh('iw', 'dev', IFACE, 'station', 'dump').stdout
|
|
|
|
# ------------------------------------------------------------ página
|
|
def iniciar_portal(self):
|
|
if self.portal:
|
|
return
|
|
wifi = self
|
|
|
|
class Portal(BaseHTTPRequestHandler):
|
|
def responder(self, codigo, corpo='', tipo='text/html; charset=utf-8', extra=None):
|
|
dados = corpo.encode('utf-8')
|
|
self.send_response(codigo)
|
|
for k, v in (extra or {}).items():
|
|
self.send_header(k, v)
|
|
self.send_header('Content-Type', tipo)
|
|
self.send_header('Content-Length', str(len(dados)))
|
|
self.send_header('Cache-Control', 'no-store')
|
|
self.end_headers()
|
|
self.wfile.write(dados)
|
|
|
|
def do_GET(self):
|
|
if self.path.split('?')[0] in ('/', '/index.html'):
|
|
return self.responder(200, wifi.pagina())
|
|
# Android/iPhone/Windows testam "tem internet?": a resposta errada faz a página abrir sozinha
|
|
return self.responder(302, extra={'Location': f'http://{AP_IP}/'})
|
|
|
|
def do_POST(self):
|
|
if self.path != '/conectar':
|
|
return self.responder(404)
|
|
n = min(int(self.headers.get('Content-Length') or 0), 4096)
|
|
campos = {k: v[0] for k, v in parse_qs(self.rfile.read(n).decode('utf-8', 'replace')).items()}
|
|
ssid = (campos.get('ssid_outra') or campos.get('ssid') or '').strip()
|
|
senha = campos.get('senha', '')
|
|
if not ssid or len(ssid) > 32 or (senha and not 8 <= len(senha) <= 63):
|
|
wifi.erro = 'Escolha a rede e use uma senha de 8 a 63 caracteres (ou deixe vazio se a rede for aberta).'
|
|
return self.responder(200, wifi.pagina())
|
|
rede = dict(ssid=ssid, senha=senha, ip_fixo=campos.get('ip_fixo') == '1')
|
|
with wifi.lock:
|
|
wifi.trocar_para = rede
|
|
self.responder(200, wifi.pagina(conectando=ssid))
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
try:
|
|
self.portal = ThreadingHTTPServer((AP_IP, 80), Portal)
|
|
except OSError as e:
|
|
log('página não subiu:', e)
|
|
return
|
|
threading.Thread(target=self.portal.serve_forever, daemon=True).start()
|
|
|
|
def parar_portal(self):
|
|
if self.portal:
|
|
self.portal.shutdown()
|
|
self.portal.server_close()
|
|
self.portal = None
|
|
|
|
def pagina(self, conectando=None):
|
|
r = self.rede()
|
|
opcoes = ''.join(
|
|
f'<label class="rede"><input type="radio" name="ssid" value="{html.escape(x["ssid"], quote=True)}"'
|
|
f'{" checked" if r and r["ssid"] == x["ssid"] else ""}><span>{html.escape(x["ssid"])}</span>'
|
|
f'<i>{"▂▄▆█"[:max(1, min(4, int((x["sinal"] + 95) / 12)))]}</i></label>' for x in self.redes)
|
|
aviso = f'<p class="erro">{html.escape(self.erro)}</p>' if self.erro else ''
|
|
if conectando:
|
|
corpo = (f'<h1>Conectando em “{html.escape(conectando)}”…</h1><p>Se der certo, a rede '
|
|
f'<b>{html.escape(self.ajustes()["ap_nome"])}</b> some e o Carinha volta ao normal em '
|
|
f'<b>http://{IP_FIXO}:8100</b>.</p><p>Se a senha estiver errada, ela volta em cerca de 1 minuto, '
|
|
'com o aviso nesta página.</p>')
|
|
else:
|
|
corpo = f"""<h1>🤖 Wi-Fi do Carinha</h1>
|
|
<p class="fraco">{'Rede salva: <b>' + html.escape(r['ssid']) + '</b> (não encontrada agora)' if r else 'Nenhuma rede salva ainda.'}</p>
|
|
{aviso}
|
|
<form method="post" action="/conectar">
|
|
<div class="redes">{opcoes or '<p class="fraco">Nenhuma rede encontrada. Digite o nome abaixo.</p>'}</div>
|
|
<label>Outra rede (nome)<input name="ssid_outra" maxlength="32" autocomplete="off"></label>
|
|
<label>Senha<input name="senha" type="password" maxlength="63" autocomplete="off"></label>
|
|
<label class="caixa"><input type="checkbox" name="ip_fixo" value="1" checked> Manter o endereço {IP_FIXO} (recomendado em casa)</label>
|
|
<button>Conectar</button>
|
|
</form>"""
|
|
return f"""<!doctype html><html lang="pt-BR"><head><meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width,initial-scale=1"><title>Wi-Fi do Carinha</title><style>
|
|
:root{{color-scheme:dark;--fundo:#0d1117;--cartao:#161b22;--borda:#30363d;--texto:#e6edf3;--fraco:#8b949e;--azul:#5cd6ff;--erro:#ff7b72}}
|
|
body{{margin:0;padding:24px 16px;background:var(--fundo);color:var(--texto);font:16px system-ui,sans-serif}}
|
|
main{{max-width:420px;margin:auto;background:var(--cartao);border:1px solid var(--borda);border-radius:16px;padding:20px}}
|
|
h1{{font-size:20px;margin:0 0 8px}} .fraco{{color:var(--fraco);font-size:14px}} .erro{{color:var(--erro)}}
|
|
form{{display:grid;gap:14px}} label{{display:grid;gap:6px;font-size:14px;color:var(--fraco)}}
|
|
input[type=password],input[name=ssid_outra]{{padding:12px;border-radius:10px;border:1px solid var(--borda);background:var(--fundo);color:var(--texto);font-size:16px}}
|
|
.redes{{display:grid;gap:6px}} .rede{{display:flex;align-items:center;gap:10px;padding:12px;border:1px solid var(--borda);border-radius:10px;color:var(--texto)}}
|
|
.rede span{{flex:1}} .rede i{{font-style:normal;color:var(--azul)}} .caixa{{display:flex;align-items:center;gap:8px}}
|
|
button{{padding:14px;border:0;border-radius:10px;background:var(--azul);color:#00222e;font-weight:700;font-size:16px}}
|
|
</style></head><body><main>{corpo}</main></body></html>"""
|
|
|
|
# ------------------------------------------------------------ laço
|
|
def rodar(self):
|
|
sh('iw', 'reg', 'set', 'BR')
|
|
rede = self.rede()
|
|
if rede:
|
|
self.modo_cliente(rede)
|
|
else:
|
|
self.modo_ap('nenhuma rede salva')
|
|
while True:
|
|
time.sleep(3)
|
|
try:
|
|
self.passo()
|
|
except Exception as e: # nunca derruba o serviço
|
|
log('erro:', e)
|
|
|
|
def passo(self):
|
|
with self.lock:
|
|
pedido, self.trocar_para = self.trocar_para, None
|
|
if pedido: # veio da página: tenta a rede nova
|
|
time.sleep(2) # deixa a resposta chegar ao celular
|
|
self.anterior = self.rede() # se a nova não funcionar, volta para esta
|
|
gravar(REDE, pedido)
|
|
self.erro = ''
|
|
self.modo_cliente(pedido)
|
|
return
|
|
agora = time.time()
|
|
if self.modo == 'cliente':
|
|
if self.conectado() and self.ip_wlan() and self.chega_na_rede():
|
|
if not self.ok_em:
|
|
log('conectado em', self.rede()['ssid'], 'com', self.ip_wlan())
|
|
if self.rede().get('ip_fixo', True):
|
|
self.liberar_ip_fixo_do_cabo()
|
|
self.erro = ''
|
|
self.anterior = False
|
|
self.ok_em = agora
|
|
self.publicar()
|
|
return
|
|
limite = QUEDA_TOLERADA if self.ok_em else ESPERA_CONECTAR
|
|
if agora - (self.ok_em or self.desde) > limite:
|
|
if not self.ok_em:
|
|
if self.conectado():
|
|
self.erro = (f'Conectei em “{self.rede()["ssid"]}”, mas por ela não chego ao roteador da casa '
|
|
f'({self.gateway_wlan() or GATEWAY}). Escolha a rede onde ficam o PC e o celular do Carinha.')
|
|
else:
|
|
self.erro = (f'Não consegui conectar em “{self.rede()["ssid"]}”. Confira a senha e se a rede está '
|
|
'ligada, e tente de novo.')
|
|
if self.anterior is not False: # a rede nova falhou: a antiga continua salva
|
|
if self.anterior:
|
|
gravar(REDE, self.anterior)
|
|
else:
|
|
REDE.unlink(missing_ok=True)
|
|
self.anterior = False
|
|
self.modo_ap('a rede não respondeu')
|
|
elif self.modo == 'ap':
|
|
r = self.rede()
|
|
if r and agora - self.desde > RETENTAR_A_CADA and not self.alguem_no_ap():
|
|
self.modo_cliente(r) # ninguém configurando: tenta a rede salva de novo
|
|
|
|
|
|
def main():
|
|
w = Wifi()
|
|
|
|
def sair(*_):
|
|
w.parar_portal()
|
|
w.parar_filhos()
|
|
raise SystemExit(0)
|
|
signal.signal(signal.SIGTERM, sair)
|
|
w.rodar()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|