#!/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 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): conteudo = f'[Match]\nName={IFACE}\n\n[Network]\nAddress={IP_FIXO}/24\nGateway={GATEWAY}\nDNS={DNS}\n' else: conteudo = f'[Match]\nName={IFACE}\n\n[Network]\nDHCP=ipv4\n\n[DHCPv4]\nRouteMetric=100\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'' for x in self.redes) aviso = f'

{html.escape(self.erro)}

' if self.erro else '' if conectando: corpo = (f'

Conectando em “{html.escape(conectando)}”…

Se der certo, a rede ' f'{html.escape(self.ajustes()["ap_nome"])} some e o Carinha volta ao normal em ' f'http://{IP_FIXO}:8100.

Se a senha estiver errada, ela volta em cerca de 1 minuto, ' 'com o aviso nesta página.

') else: corpo = f"""

🤖 Wi-Fi do Carinha

{'Rede salva: ' + html.escape(r['ssid']) + ' (não encontrada agora)' if r else 'Nenhuma rede salva ainda.'}

{aviso}
{opcoes or '

Nenhuma rede encontrada. Digite o nome abaixo.

'}
""" return f""" Wi-Fi do Carinha
{corpo}
""" # ------------------------------------------------------------ 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(): 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: 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()