Em ambientes de auditoria, registrar sessões WebSSH é essencial para rastrear ações do usuário. Ferramentas como Asciinema permitem gravação no terminal, mas integrar isso em WebSSH requer uma abordagem diferente: gerar arquivos de gravação durante a interação, sem comandos de gravação explícitos.

Arquitetura do Sistema de Gravação

O formato Asciinema utiliza um cabeçalho JSON seguido por linhas de dados de I/O. O cabeçalho, escrito no início da sessão, contém metadados como versão, dimensões e timestamp. Os dados I/O são adicionados continuamante durante a interação WebSSH.

{"version": 2, "width": 200, "height": 50, "timestamp": 1620000000.0, "env": {"TERM": "xterm-256color"}, "title": "sessao_gravada"}

Cada entrada de I/O segue a estrutura: tempo relativo, tipo (geralmente "o" para saída) e conteúdo:

[1.234, "o", "Comando executado com sucesso\r\n"]

Modelagem de Dados e Gerenciamento de Arquivos

Um modelo Django armazena metadados da sessão, associando host, usuário e arquivo de gravação.

class RegistroSessao(models.Model):
    data_criacao = models.DateTimeField(auto_now_add=True)
    host = models.ForeignKey(Host, on_delete=models.CASCADE)
    usuario = models.ForeignKey(User, on_delete=models.CASCADE)
    arquivo_gravacao = models.CharField(max_length=255)

    def __str__(self):
        return f"Sessão em {self.host} por {self.usuario}"

Durante a inicialização da conexão WebSSH, um nome de arquivo único é gerado para evitar conflitos.

def gerar_nome_arquivo(host, usuario):
    timestamp = int(time.time())
    return f"gravacao_{host}_{usuario}_{timestamp}.cast"

Lógica de Gravação em Tempo Real

Os dados são escritos em um diretório específico, com o cabeçalho gravado uma vez e os I/O anexados continuamente. Bloqueios evitam inconsistências durante escritas concorrentes.

def escrever_dados_gravacao(tipo, conteudo, caminho_arquivo, tempo_inicio):
    if tipo == 'cabecalho':
        with open(caminho_arquivo, 'w') as f:
            f.write(json.dumps(conteudo) + '\n')
    else:
        tempo_relativo = time.time() - tempo_inicio
        entrada_io = [tempo_relativo, 'o', conteudo]
        with open(caminho_arquivo, 'a') as f:
            f.write(json.dumps(entrada_io) + '\n')

Durante a sessão SSH, tanto a saída do terminal quanto erros são capturados e gravados.

def processar_dados_ssh(canal, websocket, caminho_arquivo, tempo_inicio):
    lock = threading.Lock()
    while True:
        with lock:
            dados = canal.recv(4096).decode('utf-8', errors='ignore')
            if not dados:
                break
            websocket.send(json.dumps({'status': 'sucesso', 'dados': dados}))
            escrever_dados_gravacao('iodata', dados, caminho_arquivo, tempo_inicio)

Interface de Reprodução

Uma interface web permite a reprodução das gravações usando o player Asciinema, integrado via JavaScript.

<div id="modal-player" class="modal">
  <div class="modal-content">
    <div id="player-container"></div>
  </div>
</div>

<script>
function iniciar_reproducao(arquivo) {
  const player = document.getElementById('player-container');
  player.innerHTML = `<asciinema-player src="/gravacoes/${arquivo}" speed="2" idle-time-limit="1"></asciinema-player>`;
  $('#modal-player').modal('show');
}
</script>

Considerações de Desempenho

A escolha entre gravação em tempo real e adiada envolve trade-offs. Gravação imdeiata minimiza perda de dados mas aumenta I/O. Uma abordagem intermediária pode gravar em intervalos regulares, como a cada 30 segundos, balanceando uso de memória e disco.