julioandClaude Opus 5.5
35d443a358
- comum.py: ler_json (Origin/JSON/tamanho), gravar_atomico e token (CARINHA_TOKEN ou seguranca.json), substituindo 6 copias nos modulos - eventos.py: hub SSE em /eventos com heartbeat e Last-Event-ID; modulos publicam expressao, telemetria, conversa, reacao, ia, motores - /estado agrega o snapshot inicial; /comando unifica gesto/rosto/cor/reacao - /parar-tudo desliga o PCA9685, para reacoes e trava motores ate /rearmar - braco: lease (ttl_ms) e interpolacao com limite de velocidade no servidor - braco e diagnostico nao disputam o PCA9685 (409 para o segundo dono) - fila limitada (20) com validade de 10 s; POST /face usa prefixo "rosto" - /app/config passa a checar Origin e JSON; versao vem de version.json - figurinhas automaticas do Gemini limitadas a 6 por hora - rosto entende a ordem "calma" enviada pelo E-Stop Token so e exigido quando configurado (docs/SEGURANCA.md). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
"""Peças compartilhadas pelos módulos HTTP: leitura de JSON, gravação atômica e token."""
|
|
import hmac
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
|
|
class Conflito(Exception):
|
|
"""Pedido válido, mas o estado atual não permite (vira HTTP 409)."""
|
|
|
|
|
|
def origem_valida(h):
|
|
origin = h.headers.get('Origin')
|
|
host = h.headers.get('Host', '')
|
|
return not origin or origin in ('http://' + host, 'https://' + host)
|
|
|
|
|
|
def ler_json(h, limite):
|
|
"""Corpo JSON de um POST do painel; ValueError com mensagem legível se inválido."""
|
|
if not origem_valida(h):
|
|
raise ValueError('Origem inválida')
|
|
if h.headers.get('Content-Type', '').split(';')[0] != 'application/json':
|
|
raise ValueError('Use JSON')
|
|
tamanho = int(h.headers.get('Content-Length', '0'))
|
|
if not 0 < tamanho <= limite:
|
|
raise ValueError(f'Corpo vazio ou maior que {limite} bytes')
|
|
dados = json.loads(h.rfile.read(tamanho))
|
|
if not isinstance(dados, dict):
|
|
raise ValueError('Objeto JSON obrigatório')
|
|
return dados
|
|
|
|
|
|
def gravar_atomico(caminho, obj, privado=False):
|
|
caminho = Path(caminho)
|
|
tmp = caminho.with_suffix('.tmp')
|
|
texto = json.dumps(obj, ensure_ascii=False, indent=2)
|
|
if privado:
|
|
fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, 'w', encoding='utf-8') as f:
|
|
f.write(texto)
|
|
else:
|
|
tmp.write_text(texto, encoding='utf-8')
|
|
tmp.replace(caminho)
|
|
if privado:
|
|
os.chmod(caminho, 0o600)
|
|
|
|
|
|
def carregar_token(pasta):
|
|
"""CARINHA_TOKEN do ambiente ou {"token": ...} em seguranca.json; None = sem autenticação."""
|
|
token = os.environ.get('CARINHA_TOKEN', '').strip()
|
|
if not token:
|
|
try:
|
|
token = str(json.loads((Path(pasta) / 'seguranca.json').read_text(encoding='utf-8')).get('token', '')).strip()
|
|
except (OSError, ValueError, AttributeError):
|
|
token = ''
|
|
if token and len(token) < 8:
|
|
raise ValueError('CARINHA_TOKEN precisa ter pelo menos 8 caracteres')
|
|
return token or None
|
|
|
|
|
|
def token_do_pedido(h):
|
|
auth = h.headers.get('Authorization', '')
|
|
if auth.lower().startswith('bearer '):
|
|
return auth[7:].strip()
|
|
if h.headers.get('X-Carinha-Token'):
|
|
return h.headers['X-Carinha-Token'].strip()
|
|
# EventSource e o app Android não mandam cabeçalhos próprios com facilidade.
|
|
return parse_qs(urlsplit(h.path).query).get('t', [''])[0]
|
|
|
|
|
|
def autorizado(h):
|
|
esperado = getattr(h.server, 'token', None)
|
|
if not esperado:
|
|
return True
|
|
return hmac.compare_digest(token_do_pedido(h).encode(), esperado.encode())
|