Arquitetura & Código — API UniFi Controle

1Sobre o Sistema

A API UniFi Controle Multi-Controller é um backend RESTful desenvolvido em Python, cujo propósito central é permitir o monitoramento e controle unificado de múltiplos controladores UniFi (Ubiquiti) a partir de uma única interface. O sistema foi projetado para ambientes corporativos com máquinas UniFi distribuídas em diferentes sites ou filiais.

Funcionalidades principais

Área	Funcionalidade	Descrição
Autenticação	JWT + Refresh Token Rotation	Login seguro com par access/refresh. Blacklist de tokens revogados no PostgreSQL.
RBAC	4 papéis hierárquicos	superadmin > admin > operator > readonly. Cada endpoint verifica o papel mínimo.
Multi-Controller	Gerenciar N controladores UniFi	CRUD de controladores. Senhas cifradas com Fernet AES-128. Sessão HTTP persistente por controlador.
Circuit Breaker	Proteção de controladores instáveis	Após 5 falhas consecutivas, o circuito abre por 10 min. Reseta automaticamente.
Dispositivos	Monitoramento de APs, Switches, Gateways	Lista, detalhes, reboot remoto via API UniFi.
Clientes	Monitoramento de clientes conectados	Wireless/wired, bloquear/desbloquear, paginação, filtros.
Rede	WLANs, VLANs, roteamento, status	Consulta direta ao controlador UniFi em tempo real.
Segurança	Alertas, firmware, arquivamento	Alertas de segurança com enriquecimento AP-name. Archive em lote ou individual.
Métricas	Throughput TX/RX em Mbps	InfluxDB como fonte primária; ring buffer em memória como fallback.
WebSocket	Status em tempo real	Snapshot a cada 10s. Cache compartilhado entre conexões. Limite 5 por IP.
Auditoria	Log completo de ações	Todos os eventos críticos gravados em `audit_logs`. Acesso exclusivo superadmin.
Telegram	Alertas via bot	Avisos de controlador offline, falha de login, latência alta.
Recuperação de Senha	Reset via e-mail	Token com TTL, anti-enumeração, rate-limit 5/min.
Configuração Live	Edição do .env via API	Superadmin pode alterar variáveis de sistema e reiniciar o serviço sem SSH.

2Stack Tecnológico

Categoria	Biblioteca/Ferramenta	Versão	Papel
Framework Web	`fastapi`	0.104.1	Roteamento, validação Pydantic, OpenAPI automático
ASGI Server	`uvicorn`	0.24.0	Servidor HTTP/WebSocket assíncrono
ORM	`sqlalchemy`	2.0.23	Mapeamento objeto-relacional
Migrations	`alembic`	1.13.0	Versionamento do schema do banco
DB Driver	`psycopg2-binary`	—	Conector PostgreSQL
HTTP Async	`aiohttp`	—	Comunicação com controladores UniFi
JWT	`python-jose[cryptography]`	—	Criação e verificação de tokens JWT
Senhas	`passlib[bcrypt]`	—	Hash bcrypt de senhas de usuários
Criptografia	`cryptography` (Fernet)	—	Criptografia AES-128 de senhas de controladores
Validação	`pydantic-settings`	—	Settings tipados via .env
Rate Limiting	`slowapi`	—	Limite de requisições por IP
Logging	`structlog`	—	Logs estruturados em JSON
InfluxDB	`influxdb-client`	—	Séries temporais para métricas
E-mail	`aiosmtplib`	—	Envio assíncrono de e-mails

Banco de Dados

Banco	Propósito	Tabelas / Métricas
PostgreSQL ≥ 14	Dados relacionais persistentes	users, controllers, audit_logs, revoked_tokens, password_reset_tokens, telegram_config
InfluxDB 2.0	Séries temporais de métricas	measurement: `controller_throughput` (campos: tx_mbps, rx_mbps, controller_id)

3Estrutura de Pastas

filesystemRaiz do projeto

/root/API UniFi Controle/
│
├── main.py                     # Entry point — FastAPI app, middlewares, lifespan, routers
├── requirements.txt
├── alembic.ini
├── pytest.ini
│
├── src/
│   ├── config.py               # Settings via pydantic-settings (.env)
│   ├── dependencies.py         # Dependency Injection: get_current_user, require_*
│   ├── limiter.py              # SlowAPI limiter singleton
│   │
│   ├── api/                    # Routers FastAPI (camada de apresentação)
│   │   ├── auth.py             # POST /auth/login, /auth/refresh, /auth/logout
│   │   ├── users.py            # CRUD de usuários
│   │   ├── controllers.py      # Status e dados dos controladores (read-only)
│   │   ├── controllers_crud.py # CRUD dos controladores no banco
│   │   ├── clients.py          # Clientes conectados (UniFi)
│   │   ├── device.py           # Dispositivos UniFi
│   │   ├── network.py          # WLANs, VLANs, roteamento
│   │   ├── security.py         # Alertas de segurança UniFi
│   │   ├── logs.py             # Eventos e logs do controlador
│   │   ├── audit.py            # Auditoria interna (superadmin)
│   │   ├── metrics.py          # Throughput InfluxDB / buffer
│   │   ├── telegram.py         # Configuração e teste do bot Telegram
│   │   ├── password_reset.py   # Reset de senha via e-mail
│   │   ├── env_settings.py     # Configuração live do .env (superadmin)
│   │   ├── websocket.py        # WS: status em tempo real
│   │   └── exceptions.py       # Tratadores de erros customizados
│   │
│   ├── services/               # Camada de negócio (lógica central)
│   │   ├── auth.py             # JWT, bcrypt, blacklist, CRUD de usuários
│   │   ├── database.py         # DatabaseService — CRUD controladores/usuários
│   │   ├── unifi_manager.py    # UniFiControllerClient, circuit breaker, manager
│   │   ├── audit_service.py    # Gravação e consulta de audit_logs
│   │   ├── influxdb_service.py # Conexão e escrita no InfluxDB
│   │   ├── metrics_collector.py# Background task: coleta throughput
│   │   ├── monitor_service.py  # Background task: monitora controladores (Telegram)
│   │   ├── telegram_service.py # Bot Telegram: envio de alertas
│   │   ├── smtp_service.py     # Envio de e-mails (reset de senha)
│   │   ├── password_reset_service.py # Tokens de reset + fluxo SMTP
│   │   └── throughput_buffer.py      # Ring buffer em memória (fallback InfluxDB)
│   │
│   └── models/                 # SQLAlchemy models + criação do banco
│       ├── database.py         # User, UniFiController, RevokedToken, AuditLog…
│       ├── clients.py
│       ├── device.py
│       └── network.py
│
├── alembic/                    # Migrations SQL
│   └── versions/
│       ├── 001_initial.py
│       └── 002_revoked_tokens.py
│
├── tests/
│   ├── unit/                   # 20+ arquivos de testes unitários
│   └── integration/            # 13 arquivos de testes de integração
│
├── scripts/                    # Build, install, database setup
├── debian/                     # Manifesto do pacote .deb
└── docs/                       # Esta documentação

ℹ️ Convenção de três camadas

O projeto segue api/ → services/ → models/: a camada api/ recebe requisições, delega lógica para services/, que por sua vez persiste via models/. Essa separação torna cada camada testável independentemente.

4Diagrama de Arquitetura

4.1 Visão de alto nível

┌─ CLIENTES ─────────────────────────────────────────────────────────────────┐ │ [Browser / App] [Telegram Bot] [Ferramentas CI/CD] │ └─────────────────────────┬──────────────────────────────────────────────────┘ │ HTTP/HTTPS │ WebSocket ┌─ PROXY REVERSO ──────────┼──────────────────────────────────────────────────┐ │ [Apache2 — porta 80/443] │ │ ProxyPass → http://127.0.0.1:8000 │ └─────────────────────────┬──────────────────────────────────────────────────┘ │ ┌─ API LAYER (FastAPI/Uvicorn — porta 8000) ──────────────────────────────────┐ │ │ │ Middlewares: │ │ ┌────────────────┐ ┌──────────────────┐ ┌────────────────────────────┐ │ │ │RequestIDMiddlw │ │SecurityHeadersMdl│ │ CORSMiddleware │ │ │ └────────────────┘ └──────────────────┘ └────────────────────────────┘ │ │ │ │ Routers (src/api/): │ │ ┌──────┐ ┌──────┐ ┌──────────┐ ┌────────┐ ┌────────┐ ┌───────────┐ │ │ │ auth │ │users │ │ctrls_crud│ │clients │ │ device │ │ network │ │ │ └──────┘ └──────┘ └──────────┘ └────────┘ └────────┘ └───────────┘ │ │ ┌──────────┐ ┌──────┐ ┌────────┐ ┌─────────┐ ┌──────┐ ┌─────────┐ │ │ │ security │ │ logs │ │ audit │ │ metrics │ │telegram│ │env_sett │ │ │ └──────────┘ └──────┘ └────────┘ └─────────┘ └──────┘ └─────────┘ │ │ ┌────────────────┐ ┌─────────────┐ │ │ │ password_reset │ │ websocket │ │ │ └────────────────┘ └─────────────┘ │ │ │ │ Dependency Injection (src/dependencies.py): │ │ get_current_user → require_operator → require_admin → require_superadmin │ └──────────────────────────┬───────────────────────────────────────────────┘ │ ┌─ SERVICE LAYER (src/services/) ─────────────────────────────────────────────┐ │ │ │ ┌──────────────┐ ┌───────────────┐ ┌──────────────────────────────┐ │ │ │ auth.py │ │ database.py │ │ unifi_manager.py │ │ │ │ JWT/bcrypt │ │ CRUD usuários │ │ UniFiControllerClient │ │ │ │ blacklist │ │ CRUD ctrls │ │ CircuitBreaker, Manager │ │ │ └──────────────┘ └───────────────┘ └──────────────────────────────┘ │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │audit_service │ │influxdb_svc │ │metrics_collct│ │monitor_svc │ │ │ │ fire-forget │ │ write/query │ │ background │ │ background │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ │ ┌──────────────┐ ┌───────────────┐ ┌──────────────────────────────┐ │ │ │telegram_svc │ │ smtp_service │ │ password_reset_service │ │ │ │ bot alertas │ │ email async │ │ token + SMTP │ │ │ └──────────────┘ └───────────────┘ └──────────────────────────────┘ │ └──────────────────────────┬───────────────────────────────────────────────┘ │ ┌─ DATA LAYER ────────────────────────────────────────────────────────────────┐ │ │ │ [PostgreSQL 14+] [InfluxDB 2.0] │ │ ┌──────────────────────────────┐ ┌─────────────────────────────┐ │ │ │ users │ │ measurement: │ │ │ │ controllers (pwd_encrypted) │ │ controller_throughput │ │ │ │ audit_logs │ │ fields: tx_mbps, rx_mbps │ │ │ │ revoked_tokens │ │ tag: controller_id │ │ │ │ password_reset_tokens │ └─────────────────────────────┘ │ │ │ telegram_config │ │ │ └──────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘ │ ┌─ CONTROLADORES UniFi (externos) ────────────────────────────────────────────┐ │ [Controller A] [Controller B] [Controller N] │ │ HTTPS/HTTP aiohttp — sessão persistente, circuit breaker, CookieJar unsafe │ └─────────────────────────────────────────────────────────────────────────────┘

4.2 Background Tasks (long-running)

lifespan(app) ────────────────────────────────────────────────────── │ ├─ metrics_collector.start() │ └── asyncio loop: │ ├── Para cada controlador: │ │ ├── ensure_authenticated() │ │ ├── api_request("stat/health") │ │ └── influxdb_service.write(tx_mbps, rx_mbps) │ │ └── throughput_buffer.push() (fallback) │ └── sleep(30s) │ ├─ monitor_service.start() │ └── asyncio loop: │ ├── check_all_controllers_health() │ ├── Se offline: telegram_service.alert_controller_offline() │ ├── Se latência alta: telegram_service.alert_slow() │ └── sleep(60s) │ └─ token_cleanup_loop() └── asyncio loop: ├── cleanup_expired_revoked_tokens() └── sleep(6h)

5Camadas da Aplicação

5.1 Camada de Apresentação (src/api/)

Responsável exclusivamente por receber requisições HTTP, validar schemas Pydantic, chamar services e formatar respostas. Nenhuma lógica de negócio reside aqui.

Arquivo	Prefixo	Auth mínima	Responsabilidade
`auth.py`	`/api/v1/auth`	Público*	Login, refresh token, logout
`password_reset.py`	`/api/v1/auth`	Público	Solicitar e confirmar reset de senha
`users.py`	`/api/v1/users`	JWT	CRUD de usuários, perfil próprio
`controllers_crud.py`	`/api/v1/controllers`	operator	CRUD de controladores no banco
`controllers.py`	`/api/v1/controllers`	JWT	Status, estatísticas, dados ao vivo
`device.py`	`/api/v1/device`	JWT	Dispositivos UniFi (lista, detalhe, reboot)
`clients.py`	`/api/v1/clients`	JWT	Clientes conectados, bloquear/desbloquear
`network.py`	`/api/v1/network`	JWT	WLANs, VLANs, roteamento
`security.py`	`/api/v1/security`	JWT	Alertas de segurança, firmware
`logs.py`	`/api/v1/logs`	JWT	Eventos e logs do controlador
`metrics.py`	`/api/v1/metrics`	JWT	Throughput TX/RX (InfluxDB / buffer)
`audit.py`	`/api/v1/audit`	superadmin	Logs de auditoria internos
`telegram.py`	`/api/v1/telegram`	superadmin	Configuração do bot Telegram
`env_settings.py`	`/api/v1/settings`	superadmin	Configuração live do .env
`websocket.py`	`/ws`	JWT (msg)	Status em tempo real via WebSocket

5.2 Camada de Negócio (src/services/)

Contém toda a lógica de negócio. Os services são instâncias singleton importadas pelos routers. Todos os métodos bloqueantes usam asyncio.to_thread() para não bloquear o event loop.

🔑

auth.py

src/services/auth.py

JWT (access + refresh), bcrypt, blacklist PostgreSQL com cache em memória TTL-60s, CRUD de usuários.

🗄️

database.py

src/services/database.py

DatabaseService: CRUD síncrono via asyncio.to_thread(). Descriptografa senhas dos controladores antes de retornar.

📡

unifi_manager.py

src/services/unifi_manager.py

UniFiControllerClient (aiohttp, sessão persistente, circuit breaker 5 falhas/10 min). ControllersManager gerencia um dict de clientes.

📝

audit_service.py

src/services/audit_service.py

Fire-and-forget via asyncio.create_task(). Insere em audit_logs sem bloquear a requisição. Erros são silenciosos.

📊

influxdb_service.py

src/services/influxdb_service.py

Abstrai conexão e escrita/consulta no InfluxDB 2.0. Resiliência: não bloqueia startup se InfluxDB estiver offline.

⏱️

metrics_collector.py

src/services/metrics_collector.py

Background task: coleta TX/RX de todos os controladores a cada 30s. Escreve no InfluxDB e no ThroughputBuffer.

🛡️

monitor_service.py

src/services/monitor_service.py

Background task: detecta controladores offline ou lentos a cada 60s. Dispara alertas Telegram com cooldown configurável.

📲

telegram_service.py

src/services/telegram_service.py

Envio de mensagens via Bot API Telegram. Suporta alertas de login, offline, latência. Token mascarado na API.

5.3 Camada de Dados (src/models/)

Model SQLAlchemy	Tabela	Campos relevantes
`User`	users	id, username, hashed_password, full_name, email, role (enum), disabled, created_at, last_login
`UniFiController`	controllers	id, name, host, port, username, password_encrypted (Fernet), verify_ssl, site, enabled, last_seen
`RevokedToken`	revoked_tokens	id, token_hash (SHA-256), expires_at — limpeza periódica a cada 6h
`AuditLog`	audit_logs	id, username, action, user_id, role, resource, detail, ip_address, status, duration_ms, created_at
`PasswordResetToken`	password_reset_tokens	id, user_id, token_hash, expires_at, used
`TelegramConfig`	telegram_config	bot_token, chat_id, enabled, thresholds, cooldown_minutes

6Fluxo de Dados — Requisição Típica

# Exemplo: GET /api/v1/clients/3?connection_type=wireless [Browser] ──1─► [Apache2:80] ProxyPass ──2─► [Uvicorn:8000] ──3─► RequestIDMiddleware gera X-Request-ID: uuid4 ──4─► SecurityHeadersMiddleware injeta 5 headers de segurança ──5─► CORSMiddleware verifica Origin ──6─► get_current_user() valida Bearer JWT │ ├── _get_cached_user(username) → hit? retorna em O(1) │ └── miss: database_service.get_user_by_username() │ → asyncio.to_thread(_sync_db_query) ──7─► clients.get_clients() handler │ ├── _get_uc(controller_id) → controllers_manager.clients[3] │ ├── uc.ensure_authenticated() │ │ ├── _circuit_is_open? → False → prosseguir │ │ └── needs_login? → login() via aiohttp POST /api/login │ ├── uc.get_clients() → aiohttp GET /api/s/default/stat/sta │ ├── [_unifi_client_to_dict(c) for c in raw] │ └── filtrar por connection_type + paginação ◄─8── [Browser] JSON response + X-Request-ID

6.1 Fluxo de Escrita — Criar Controlador

# POST /api/v1/controllers/ (require_admin) [Admin] ──► controllers_crud.create_controller() ──► database_service.add_controller(password=...) ├── crypto.encrypt_password(password) ← Fernet AES-128 ├── session.add(UniFiController(password_encrypted=...)) ├── session.commit() └── retorna { success: true, controller: {...} } ──► controllers_manager.reload_controllers() ├── DELETE clientes antigos da memória ├── database_service.get_controllers(enabled_only=True) │ └── decrypt_password para cada controller └── cria novo UniFiControllerClient para cada registro ◄── [Admin] 201 Created + controlador criado

7Fluxo de Autenticação

7.1 Login e emissão de tokens

# POST /api/v1/auth/login rate-limit: 10/min por IP [Cliente] ──► { username, password } │ ├─ authenticate_user(username, password) │ ├── get_user_by_username(username) ← PostgreSQL │ └── pwd_context.verify(password, hash) ← bcrypt timing-safe │ ├─ user.disabled? ──► 403 Forbidden │ ├─ create_access_token({ sub: username, role: role }, expires=30min) │ └── jwt.encode(payload, SECRET_KEY, algorithm=HS256) │ ├─ create_refresh_token({ sub: username }, expires=7d) │ ├─ audit_service.log(action="login") ← fire-and-forget ├─ database_service.update_user(last_login=now) │ └─ 200 { access_token, refresh_token, expires_in, user } ───────────────────────────────────────────────────────────────── # Uso do access_token em rotas protegidas Authorization: Bearer <access_token> └─ get_current_user() ├── HTTPBearer extrai token ├── verify_token(token): │ ├── jwt.decode() → payload │ ├── _get_cached_user(sub) → TTL 60s │ └── miss: get_user_by_username() + _set_cached_user() └── retorna User object ───────────────────────────────────────────────────────────────── # POST /api/v1/auth/refresh (Refresh Token Rotation) { refresh_token } ├─ verify_refresh_token(token) │ ├── jwt.decode() → { sub } │ ├── _is_revoked_sync(SHA-256(token)) ← fail-CLOSED │ └── get_user_by_username(sub) │ ├─ async_revoke_refresh_token(old_token) ← persistir em revoked_tokens │ └─ { new_access_token, new_refresh_token } # POST /api/v1/auth/logout └─ async_revoke_refresh_token(token) → INSERT revoked_tokens

7.2 Hierarquia RBAC

texto

superadmin  ──► Gestão de usuários, configurações do sistema, auditoria, Telegram
    │
    admin   ──► CRUD de controladores, criar/alterar usuários (exceto superadmin)
       │
     operator ──► Reboot de dispositivos, bloquear clientes, leitura geral
          │
        readonly ──► Apenas leitura (clientes, dispositivos, rede)

ℹ️ Implementação

A hierarquia está centralizada em src/dependencies.py como frozensets. A dependência require_admin inclui todos os roles superiores ao admin. Nenhuma regra de role está duplicada nos routers.

8Fluxo WebSocket

# WS /ws/status [Cliente] ──WS CONNECT──► Uvicorn │ ├─ Limite por IP (max 5): _acquire_ws_slot(ip) ├─ websocket.accept() ├─ Aguarda mensagem JSON: { "token": "<JWT>" } ← timeout 10s ├─ verify_token(token) ← autentica │ └─ Loop a cada 10s: ├─ _controllers_snapshot_cached() │ ├─ cache TTL: 8s ← apenas 1ª conexão faz chamada ao UniFi │ └─ retorna snapshot de todos os controladores │ (id, name, host, online, wlan_clients, alarms) └─ websocket.send_text(json) Desconexão: WebSocketDisconnect └─ _release_ws_slot(ip)

ℹ️ Design pattern: Shared Snapshot Cache

O cache compartilhado entre conexões WS evita que N clientes conectados gerem N×K (controladores) chamadas ao UniFi por ciclo. Com TTL de 8s e intervalo de envio de 10s, sempre há dados frescos sem carga excessiva nos controladores.

9Módulo: main.py

Entry point da aplicação. Define a instância FastAPI, middlewares, routers e o ciclo de vida (lifespan).

Middlewares aplicados (ordem de execução)

#	Middleware	Tipo	Função
1	`RequestIDMiddleware`	ASGI puro	Injeta `X-Request-ID: uuid4` em toda resposta. Facilita correlação de logs.
2	`SecurityHeadersMiddleware`	ASGI puro	Adiciona: X-Content-Type-Options, X-Frame-Options, X-XSS-Protection, Referrer-Policy, Permissions-Policy.
3	`CORSMiddleware`	Starlette	Permite origens definidas em `ALLOWED_ORIGINS`. Nunca usa wildcard `*`.
4	`SlowAPI`	Rate Limiter	Limita requisições por IP. Login: 10/min. Reset de senha: 5/min.

ℹ️ ASGI puro vs. BaseHTTPMiddleware

RequestIDMiddleware e SecurityHeadersMiddleware foram implementados como ASGI puro (__call__(scope, receive, send)) para evitar o bug anyio.EndOfStream que ocorre com BaseHTTPMiddleware em cenários de streaming/WebSocket.

Lifespan — Ordem de inicialização

Fase	Ação	Comportamento em falha
1	cleanup_expired_revoked_tokens()	Warning — não bloqueia
2	ensure_default_admin()	Warning — não bloqueia
3	controllers_manager.load_controllers()	Warning — não bloqueia
4	influxdb_service.connect()	Warning — não bloqueia
5	metrics_collector.start()	Background task — falhas isoladas
6	token_cleanup_loop (task)	Background task
7	monitor_service.start() (task)	Background task

Documentação protegida

Os endpoints /docs e /openapi.json são protegidos por JWT via query param ?token=<jwt>. O token é removido do histórico do browser imediatamente via history.replaceState(). O Swagger UI padrão é desabilitado (docs_url=None), substituído por ReDoc customizado com tema dark.

10Módulo: config.py

Define a classe Settings usando pydantic-settings. Carrega variáveis de /etc/api-unifi-controle/.env (produção) ou .env local (desenvolvimento). Suporta os candidatos: CONFIG_PATH env var → /opt/unificontrole/config/.env → config/.env → .env.

Validators de segurança

Em modo produção (debug=False), a classe valida:

secret_key não pode ser o valor padrão de desenvolvimento
encryption_key deve estar presente e ter formato Fernet válido
Lança ValueError se a validação falhar, impedindo o startup

⚠️ Ponto de atenção

A lista de candidatos ainda inclui /opt/unificontrole/config/.env (path legado). No ambiente .deb, o path correto é /etc/api-unifi-controle/.env via variável CONFIG_PATH no systemd.

11Módulo: services/auth.py

Funções e variáveis importantes

Símbolo	Tipo	Descrição
`pwd_context`	CryptContext	bcrypt com auto-upgrade. Usado para hash e verify de senhas.
`_revoked_cache`	Set[str]	Cache em memória de hashes SHA-256 de tokens revogados. Complementa a tabela PostgreSQL.
`_user_cache`	Dict[str, Tuple]	Cache de objetos User com TTL de 60s. Evita query ao banco em cada requisição autenticada.
`_hash_token()`	func	SHA-256 hex do token — nunca armazena texto claro na blacklist.
`_is_revoked_sync()`	func	Fail-CLOSED: se o banco falhar, assume token revogado. Segurança > disponibilidade.
`verify_token()`	async	Decodifica JWT, verifica blacklist, retorna User. Usado por get_current_user.
`ensure_default_admin()`	async	Cria usuário superadmin padrão se a tabela users estiver vazia. Chamado no lifespan.

Política de Refresh Token Rotation

Ao usar /auth/refresh, o token antigo é imediatamente revogado (inserido em revoked_tokens) e um novo par é emitido. Isso previne reutilização de refresh tokens roubados (Fix #6 do changelog interno). O token é armazenado apenas como hash SHA-256.

12Módulo: services/unifi_manager.py

Circuit Breaker

Parâmetro	Valor padrão	Configurável?
Threshold de falhas	5 consecutivas	Constante de classe
Tempo de reset	10 minutos	Constante de classe
Reuso de sessão	55 minutos	Constante de classe

UniFiControllerClient — métodos principais

Método	Descrição
`ensure_authenticated()`	Verifica circuit breaker; reautentica se sessão expirou (>55min)
`login()`	POST /api/login com CookieJar unsafe (necessário para IPs)
`api_request(endpoint, method, data)`	Wrapper genérico para chamadas à API UniFi. Formato: GET /proxy/network/api/s/{site}/{endpoint}
`get_devices()`	stat/device
`get_clients()`	stat/sta

ℹ️ CookieJar unsafe

O aiohttp.CookieJar(unsafe=True) é necessário para aceitar cookies de endereços IP (ex.: 192.168.1.1). Por padrão, o aiohttp rejeita cookies de IPs, causando 401 em todas as requisições pós-login.

13Padrões Utilizados

Padrão	Onde	Descrição
Layered Architecture	api/ → services/ → models/	Separação clara entre apresentação, lógica e dados
Dependency Injection	FastAPI Depends()	get_current_user injetado em todos os routers protegidos
Singleton Service	database_service, audit_service	Instâncias únicas importadas pelos routers
Circuit Breaker	UniFiControllerClient	Protege requisições a controladores instáveis
Fail-Closed	_is_revoked_sync()	BD indisponível → token considerado revogado
Fire-and-Forget	audit_service.log()	asyncio.create_task() para não bloquear a resposta
Token Rotation	auth/refresh	Refresh token antigo revogado a cada novo refresh
Anti-Enumeration	password_reset/request	Sempre retorna 200 independente do e-mail existir
Cache TTL	_user_cache, _snapshot_cache	Evita queries repetidas; TTLs conservadores (60s, 8s)
Ring Buffer	ThroughputBuffer	Fallback in-memory para throughput quando InfluxDB está offline
Shared Snapshot	WebSocket	Cache único para N clientes WS — evita sobrecarga nos controladores
Data Normalization	clients.py, device.py, logs.py	Funções _unifi_*_to_dict() normalizam diferentes formatos da API UniFi
RBAC Centralizado	dependencies.py	Hierarquia de roles em frozensets — única fonte da verdade

14Análise de Segurança

Pontos fortes de segurança

Mecanismo	Implementação	Nível
Hash de senhas	bcrypt, rounds automáticos (passlib)	✔ FORTE
JWT	HS256, access 30min + refresh 7d com rotation	✔ FORTE
Senhas de controladores	Fernet AES-128 (chave no .env)	✔ FORTE
Token blacklist	SHA-256 no PostgreSQL, fail-CLOSED	✔ FORTE
Headers de segurança	ASGI middleware em todas as respostas	✔ FORTE
CORS	Lista explícita, nunca wildcard *	✔ FORTE
Rate Limiting	SlowAPI: login 10/min, reset senha 5/min	✔ BOM
Anti-Enumeration	password reset sempre retorna 200	✔ FORTE
Docs protegidos	/docs e /openapi.json exigem JWT	✔ FORTE
Config sensível	.env modo 0600, dono unifi-api	✔ FORTE
WS Auth	Token via 1ª mensagem (não query string)	✔ FORTE

Pontos de atenção

Item	Risco	Mitigação sugerida
HS256 (simétrico)	MÉDIO	Considerar RS256 em cenários multi-serviço
Blacklist por processo	MÉDIO	Com múltiplos workers, token revogado em um worker pode não ser bloqueado por outro no mesmo segundo. Falha mitigada pelo PostgreSQL.
SSL desabilitado nos controladores	MÉDIO	verify_ssl=False é o padrão. Em redes não confiáveis, usar verify_ssl=True + cert válido.
Usuário cache TTL 60s	BAIXO	Usuário desabilitado pode usar a API por até 60s. Chamar invalidate_user_cache() ao desabilitar.
Ausência de 2FA	MÉDIO	Implementar TOTP (pyotp) para superadmin
env_settings: reinício direto	MÉDIO	Superadmin pode reiniciar o serviço pela API. Adicionar confirmação de senha antes do restart.

15Riscos, Melhorias e Refatorações

Riscos identificados

Item	Risco	Impacto	Sugestão
Sessões aiohttp não fechadas explicitamente	BAIXO	Memory leak em caso de muitos controladores inativos	Cleanup no lifespan: fechar sessões de clientes desabilitados
_user_cache não thread-safe	BAIXO	Raro com asyncio FIFO, mas possível com múltiplos workers	asyncio.Lock() ao atualizar / ler o cache
ThroughputBuffer sem persistência	BAIXO	Restart do serviço limpa histórico em memória	Já resolvido pelo InfluxDB como primário
config.py path legado /opt/unificontrole/	BAIXO	Pode carregar .env incorreto em ambientes mistos	Remover candidate legado, manter apenas CONFIG_PATH e .env
Ausência de pagination no audit_logs query	MÉDIO	Em produção com muitos logs, query pode ser lenta	Já implementado via page/page_size, garantir índice em created_at + action
Sem teste de integração para WebSocket	MÉDIO	Regressões no WS podem não ser detectadas	Adicionar testes com httpx + anyio WebSocket client

Sugestões de melhoria técnica

Adicionar índices compostos em audit_logs(username, created_at) e revoked_tokens(token_hash, expires_at)
Extrair PasswordPolicy para injeção de dependência em vez de regex duplicada em dois arquivos
Adicionar OpenTelemetry tracing para correlação de requisições distribuídas
Implementar graceful shutdown mais explícito: aguardar tarefas pendentes ao encerrar
Adicionar health probe separado para liveness vs. readiness (Kubernetes/systemd)
Substituir asyncio.to_thread() por SQLAlchemy async AsyncSession para eliminar threads de DB
Implementar pagination cursor-based para os endpoints de logs e audit com volumes grandes

16Relatório Final — Avaliação do Código

Resumo executivo

O projeto é uma API REST de qualidade produção-ready com arquitetura em três camadas bem definidas, segurança acima da média para projetos FastAPI, ampla cobertura de funcionalidades para o domínio UniFi e um conjunto robusto de mecanismos de resiliência. A base de código demonstra evolução iterativa com fixes documentados (Fix #1 a #23), o que evidencia um processo de melhoria contínua orientado por problemas reais.

Pontos fortes

Categoria	Avaliação	Justificativa
Segurança	A	bcrypt, Fernet, JWT rotation, blacklist PostgreSQL, headers OWASP, fail-CLOSED, anti-enumeration
Arquitetura	A-	3 camadas claras, DI via FastAPI Depends, RBAC centralizado, padrões bem estabelecidos
Resiliência	B+	Circuit breaker, fallback buffer, graceful startup (falhas não bloqueiam), fire-and-forget para auditoria
Manutenibilidade	B+	Código documentado (docstrings), nomes descritivos, funções pequenas e focadas
Performance	B	asyncio correto, cache TTL, snapshot WS compartilhado; SQLAlchemy síncrono (to_thread) é o maior gargalo
Testabilidade	B-	20+ testes unitários, 13 integração; ausência de WS e alguns cenários edge-case
Observabilidade	B	structlog JSON, auditoria completa, health endpoint, X-Request-ID; falta tracing distribuído

Pontos fracos

SQLAlchemy síncrono com asyncio.to_thread() — funciona, mas não é ideal para alta concorrência
Cache de usuário não invalida em todos os cenários (ex.: update de role por outro worker)
Path legado /opt/unificontrole/ ainda presente em config.py
Regex de validação de senha duplicada em users.py e password_reset.py
Ausência de paginação cursor-based para grandes volumes de auditoria

✅ Conclusão

O código-base é sólido, seguro e bem estruturado para um projeto de gestão de infraestrutura. As melhorias sugeridas são incrementais e não comprometem a operação atual. Prioridade recomendada: (1) índices de banco, (2) SQLAlchemy async, (3) invalidação de cache cross-worker, (4) testes de WebSocket.

Arquitetura & Documentação Técnica