REA (Reverse Engineering Anything). Análisis

Introducción

Créditos:

1. Melina Blanco
2. Jorge Marfil y Jorge Marfil

La ingeniería inversa es muy util cuando quieres migrar de lenguaje, por ejmplo pasar código de Visual Basic 6 o Java  a Python. Pero a veces por el motivo que fuere, no se dispone de las fuentes originales. 

Ahí es donde entra en acción REA que según dicen es capaza de obtener código fuente de un lenguaje  de alto nivel a partir de código ejecutable.

Hay herramientas de ingeniería inversa que són las que llama REA:

• Ghidra que es gratis y desarrollada por la NSA (National Security Agence) de 2019 c0n licencia Apache.
• Hopper para Mac
• Ida Pro que es la versión cara que va desde 1100 € a 8600 € por año.

El principal escollo de las herramientas anteriores es que son engorrosas de usar.

 Un problema que tiene estas herramienta es que a veces no puede encontrar el nombre real de las funciones y las llama "funct" mas la posicón de memoria. A veces es casi irreconocible el código que devuelve pues es casi ensamblador. Por eso REA intentar aprovechar este código y convertirlo al lengiaje de alto nivel deseado. Y una vez puesto, que sea compatible en la mayoría de sistemas operativos

Puede haber algún problema de patentes si intentas analizar y replicar software ya existente. Todo depende del nivel que entres. Pero las leyes cambian y seguramente a partir de este proyecto viral, de seguro van a haber cambios en la legislación.

¿Que és REA?

Es un servidor Open Source que utiliza un software MCP (Model Control Protocol) que es un conector de interoperabilidad entre diferentes herramientas por ejemplo decompiladores, debuggers, inspectores web .. ,con agentes de IA (Claude Code, Codex, Cursor) para generar ingeniería inversa y por tanto la obtención de código fuente o versión nuestra de la aplicación que le hemos pasado.

Imagen de Melina Blanco

Pero REA integra aplicaciones de obtencion de pseudocódigo (Ghidra, Frida, JEB, Hopper, Playwright y ADP)

Si le pides que te analice una aplicación te explica 3 cosas sin inventarse nada, es decir te dice cuales han sido sus límites:

1. Como funciona
2. De donde sacó cada elemento (Evidencia)
3. Lo que no ha podido averiguar.

El código fuente original no te lo puede dar, pero si puede darte un esquema de las funciones que entran en juego. Y a continuación el agente de IA, basándose en ese exquema crea un código fuente y lo pureba y va ajustándolo con tu ayuda.

En el readme, se ve lo que puede leer:

Al analizar el programa (entre las opciones de la tabla anterior, se devuelve:

1. Pseudocódigo parecido a C (a partir del códigp máquina)
2. A veces JS y módulos
3. Ensamblador (instruciones del procesador)
4. Mapas de módulos
5. Rutas de llamadas
6. Actividad de red del programa.

Y ahora el agentge escribe y prueba una implementación.

Prerrequisitos

Cloud code: En Ubuntu ejecutar:

curl -fsSL https://claude.ai/install.sh | bash

Y ahora comprobar si funciona:

claude --version

Y me ha respondido

2.1.303 (Claude Code)

Y aquí tenemos un problema, y es que no podemos utilizar Claude Code sin tener una subscripción de pago

¿Como instalarlo?

Para ello:

1. Melina ha creado una carpeta para este proyecto y la añade utilizando Claude Code
2. Escribe este propmt a Claude Code: "Copia este repo e instala REA:  https://github.com/morluto/rea   
3. Lo que hace es clonar el repositorio en esa carpeta para poder ser utilizado.
4. Cuando acaba de instalarlo aparece una especio de log  donde dice donde lo ha instalado, que comando utilizar para ejecutarlo. También indica que programas se requieren en su caso lo dice que le falta Hopper o GLH.
5. Pide reiniciar sesión de Claude Code para que surta efectos la nueva configuración.

Para Android hace falta instalar JADX y JDK

Ahora Melina le pide que copie y cree una nueva versión de la aplicación Loom que usa ella, copiando lo más básico que pueda, añadiéndole un toque personal de camaleones que es su empresa y que lo meta dentro de una carpeta.

Para ello utiliza el model Opus 5.5 con esfuerzo Medio pues considera que es suificiente y no quiere que se gasten muchis tokens. Ha tardado alrededor de 6 minutos.

¿Se puede trabajar igual sin REA?

Para aquellas personas más versadas en ingeniería inversa, tal vez con Claude Code se puede pedir que jonstale Ghidra y que a partir de ahí haga lo mismo. Pero para las personas que no tienen una metodología clara de actuar es mejor iniciarse en este proceso con REA.

Cuidadín!

Jorge Marfil nos da una serie de consejos para no tener problemas. Lo que ha hecho es hacer una aplicación que le pasas un código y te deja pasar o no en base a una serie de premisas. Y para ofuscar mas la aplicación a pasado un programa para no dar ningún nombnre a las funcionas para no dar pistas a los programas de ingeniería inversa sobre el proceso de validación.

A continuación le pasa un agente que llama a REA (MCP) y este a un programa de ingenieria inversa y le informa que tiene 242 funciones nuevas sin nombre explícito como ya se había acordado.

Cada HALLAZGO, viene con su procedencia, con el porcentaje de seguridad del modelo y lo que no ha conseguido averiguar. Que se resume en tres palabras: código, referencias e incógnitas

Posibles problemas:

1. Se pueden abrir binarios de los que no se sabe nada.
2. Dentro de cada binario puede haber instrucciones ocultas para que las lea el agente y las tome por órdenes que las puede obedecer.
3. Para ello se recomienda ejecutarlo en un "sandbox" o entorno cerrado donde si algo sale mal, no afecte a nada fuera de él.

Hay que tenr bien acotado el programa respecto a su comportamiento. Así se puede testear correctamente los resultados. Es decir que sea comprobable!

El funcionamiento de REA es que le pide en este caso a Ghida que obtenga solo algunas funciones, descartando las demás. Es decir REA es capaza de buscar solo las llamadas que hacen falta.

Descargar vídeos que pueden desaparecer. Utiliza Ubuntu

Según  Claude:

Preparación (solo una vez)

1. Instala las herramientas

sudo apt install ffmpeg
curl -fsSL https://deno.land/install.sh | sh
sudo curl -L https://github.com/yt-dlp/yt-dlp/releases/latest/download/yt-dlp -o /usr/local/bin/yt-dlp
sudo chmod a+rx /usr/local/bin/yt-dlp



Si tenías yt-dlp instalado con pip, desinstálalo antes (pip uninstall yt-dlp). El binario independiente evita el aviso de Python 3.10.

2. Guarda la configuración para no repetirla

mkdir -p ~/.config/yt-dlp
echo "--remote-components ejs:github" >> ~/.config/yt-dlp/config
echo "-S ext:mp4:m4a" >> ~/.config/yt-dlp/config

A partir de ahora esas dos opciones se aplican solas.

3. Mantén yt-dlp actualizado

yt-dlp -U


Ejecútalo si algún día deja de funcionar, porque YouTube cambia a menudo.

1. Vídeo de Blogger (que no es tuyo)

Los blogs suelen incrustar vídeos de YouTube, aunque a veces usan el alojamiento propio de Blogger.

• Si es un vídeo de YouTube incrustado: pasa la URL de la página del blog o, mejor, la URL directa del vídeo de YouTube (clic derecho sobre el reproductor → Copiar URL del vídeo).
• Si es un vídeo alojado en Blogger: pasa la URL de la página del blog. Si no lo detecta, abre las herramientas de desarrollador del navegador (F12 → Elements), busca el <iframe> del vídeo y copia su src (algo como https://www.blogger.com/video.g?token=...).

yt-dlp --download-sections "*00:00:05-00:00:15" -o "%(title)s.%(ext)s" "URL"

No uses cookies: no hacen falta para contenido público.

2. Vídeo de YouTube (que no es tuyo)

yt-dlp --download-sections "*00:00:05-00:00:15" -o "%(title)s.%(ext)s" "https://youtu.be/XXXXXXXXXXX"

• El formato del tiempo es *inicio-fin (*1:30-3:45 también vale).
• Para un corte más preciso, añade --force-keyframes-at-cuts (es más lento).
• Para solo el audio: añade -x --audio-format mp3.
• No uses cookies: no las necesitas y así evitas que YouTube marque tu cuenta.

3. Tu canal, vídeos no públicos

Depende de la visibilidad:

• No listado: funciona igual que en el punto 2, sin cookies.
• Privado: necesita las cookies de la sesión iniciada en tu navegador:

yt-dlp --cookies-from-browser chrome --download-sections "*00:00:05-00:00:15" -o "%(title)s.%(ext)s" "https://youtu.be/XXXXXXXXXXX"


Cambia chrome por firefox, chromium, etc., según el navegador donde has iniciado sesión con la cuenta propietaria. Usa cookies solo para esto, no en todos los comandos.

Alternativa sin cookies: descarga el original desde YouTube Studio (Contenido → ⋮ → Descargar) y recorta con ffmpeg:

ffmpeg -ss 00:00:05 -to 00:00:15 -i "video.mp4" -c copy recorte.mp4

Si algo falla

• "n challenge solving failed" o "page needs to be reloaded": actualiza yt-dlp y comprueba que deno funciona (deno --version).
• Vídeos que piden iniciar sesión sin ser privados: en contenido ajeno no fuerces el uso de cookies; prueba primero a actualizar.
• El archivo queda en la carpeta actual. Compruébalo con pwd, o usa -o "~/Vídeos/%(title)s.%(ext)s" para elegir la carpeta.

Un aviso importante

Para los puntos 1 y 2, descargar vídeos ajenos va contra las condiciones de uso de YouTube y el contenido suele tener derechos de autor. Es razonable para uso personal, contenido con licencia libre (Creative Commons, dominio público) o con permiso del autor. Si vas a publicar o reutilizar el tramo, necesitas la autorización del titular o que encaje en una excepción legal, como la cita. Eso varía según el país, así que conviene comprobarlo.

AYTOS-FIRMADOC-FACE: (4) Llamada a Web Services

 1. Obtención del wsdl

Para ello nos conectamos con un navegador a esta url:

http://servidor_accede:9080/services/Ci?wsdl

siendo:

• servidor_accede la ip del servidor AYTOS2-ACCEDE por ejemplo 192.168.xxx.xxx
• http (no utiliza https)

En Chrome de Ubuntu me sale la página en blanco, pero en Mozilla sale esto:

<wsdl:definitions targetNamespace="http://192.168.xxx.xxx:9080/services/Ci">
<!--
WSDL created by Apache Axis version: #axisVersion#
Built on #today#
-->
<wsdl:types>
<schema targetNamespace="http://gestion.util.aytos">
<import namespace="http://schemas.xmlsoap.org/soap/encoding/"/>
<complexType name="Traza">
<sequence>
<element name="aplicacion" nillable="true" type="soapenc:string"/>
<element name="ent" nillable="true" type="soapenc:string"/>
<element name="fichero" nillable="true" type="soapenc:string"/>
<element name="idLineas" nillable="true" type="soapenc:string"/>
<element name="modulo" nillable="true" type="soapenc:string"/>
<element name="nivel" type="xsd:int"/>
<element name="org" nillable="true" type="soapenc:string"/>
<element name="serv" nillable="true" type="soapenc:string"/>
<element name="usuario" nillable="true" type="soapenc:string"/>
</sequence>
</complexType>
</schema>
</wsdl:types>
<wsdl:message name="servicioRequest">
<wsdl:part name="in0" type="soapenc:string"> </wsdl:part>
<wsdl:part name="in1" type="tns1:Traza"> </wsdl:part>
</wsdl:message>
<wsdl:message name="servicioAliasRequest">
<wsdl:part name="in0" type="soapenc:string"> </wsdl:part>
</wsdl:message>
<wsdl:message name="servicioResponse1">
<wsdl:part name="servicioReturn" type="soapenc:string"> </wsdl:part>
</wsdl:message>
<wsdl:message name="servicioAliasResponse">
<wsdl:part name="servicioAliasReturn" type="soapenc:string"> </wsdl:part>
</wsdl:message>
<wsdl:message name="servicioRequest1">
<wsdl:part name="in0" type="soapenc:string"> </wsdl:part>
</wsdl:message>
<wsdl:message name="servicioResponse">
<wsdl:part name="servicioReturn" type="soapenc:string"> </wsdl:part>
</wsdl:message>
<wsdl:portType name="Ci">
<wsdl:operation name="servicio" parameterOrder="in0 in1">
<wsdl:input message="impl:servicioRequest" name="servicioRequest"> </wsdl:input>
<wsdl:output message="impl:servicioResponse" name="servicioResponse"> </wsdl:output>
</wsdl:operation>
<wsdl:operation name="servicio" parameterOrder="in0">
<wsdl:input message="impl:servicioRequest1" name="servicioRequest1"> </wsdl:input>
<wsdl:output message="impl:servicioResponse1" name="servicioResponse1"> </wsdl:output>
</wsdl:operation>
<wsdl:operation name="servicioAlias" parameterOrder="in0">
<wsdl:input message="impl:servicioAliasRequest" name="servicioAliasRequest"> </wsdl:input>
<wsdl:output message="impl:servicioAliasResponse" name="servicioAliasResponse"> </wsdl:output>
</wsdl:operation>
</wsdl:portType>
<wsdl:binding name="CiSoapBinding" type="impl:Ci">
<wsdlsoap:binding style="rpc" transport="http://schemas.xmlsoap.org/soap/http"/>
<wsdl:operation name="servicio">
<wsdlsoap:operation soapAction=""/>
<wsdl:input name="servicioRequest">
<wsdlsoap:body encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" namespace="http://ci.sw.aytos" use="encoded"/>
</wsdl:input>
<wsdl:output name="servicioResponse">
<wsdlsoap:body encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" namespace="http://192.168.28.136:9080/services/Ci" use="encoded"/>
</wsdl:output>
</wsdl:operation>
<wsdl:operation name="servicio">
<wsdlsoap:operation soapAction=""/>
<wsdl:input name="servicioRequest1">
<wsdlsoap:body encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" namespace="http://ci.sw.aytos" use="encoded"/>
</wsdl:input>
<wsdl:output name="servicioResponse1">
<wsdlsoap:body encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" namespace="http://192.168.28.136:9080/services/Ci" use="encoded"/>
</wsdl:output>
</wsdl:operation>
<wsdl:operation name="servicioAlias">
<wsdlsoap:operation soapAction=""/>
<wsdl:input name="servicioAliasRequest">
<wsdlsoap:body encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" namespace="http://ci.sw.aytos" use="encoded"/>
</wsdl:input>
<wsdl:output name="servicioAliasResponse">
<wsdlsoap:body encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" namespace="http://192.168.28.136:9080/services/Ci" use="encoded"/>
</wsdl:output>
</wsdl:operation>
</wsdl:binding>
<wsdl:service name="CiService">
<wsdl:port binding="impl:CiSoapBinding" name="Ci">
<wsdlsoap:address location="http://192.168.xxx.xxx:9080/services/Ci"/>
</wsdl:port>
</wsdl:service>
</wsdl:definitions>

Ahora falta ver como se llaman a las operaciones

AYTOS-FIRMADOC-FACE: (3)La aventura del compresor de ficheros de Firmadoc de Aytos (2). Versión definitiva

 0. Introducción

Esto parece una película. Primeramente busqué en Google "Algoritmo de Huffman en Visual Basic" y me hace referencia a esta URL de jrubi  que ha encontrado el algoritmo de Huffman realizado por James Vincent Carnicelli en agosto del 2000, y le pasé el código a Claude para que me lo convirtiera a Python.

Como Claude tenía ya muestras de ficheros originales y comprimidos, me contesta que el algoritmo casi daba resultados correctos pero que era una versión antigua, en vez de darle cabecera "HF3" daba "HF2", y me propone buscar el algoritmo más reciente en internet y le digo que si.

Entonces encuentra esta URL de post.bytes  que dice que es :"Es un hilo de 2005 (espejo de un antiguo grupo de noticias de Microsoft sobre VB.NET) donde un usuario llamado "Crouchie1998" responde a una pregunta pegando el código VB6 completo de la clase clsHuffman, atribuido originalmente a David Midkiff. Ese mismo mensaje también enlaza a dos copias del programa VB6 ya compilado (Planet Source Code y a1vbcode.com), aunque esos enlaces concretos ya no funcionan hoy en día — el texto del código en sí es lo que pude usar."

Y primeramente prueba el algoritmo en VB.NET sobre los ficheros que tenía y me suelta "BINGO!" y me genera el algoritmo en Python completamente funcional.

Con esto conseguimos no utilizar código que puede ser propiedad intelectual de Aytos como compress.dll (aunque el algoritmo subyacente no sea de su propiedad) y poder trabajar perfectamente en Linux para obtener estos ficheros y tener un histórico documental

1. Algoritmo de Huffman de compresión y decompresión

Veamos pues el preciado algoritmo pero teniendo en cuenta que:

1. En la función main_edu() hay que darle la relació de ficheros a descomprimir
2. Los nombres de ficheros serán del tipo "12345.EXT.bin" donde "12345" representa el campo IDX_ID de la tabla BIB_AY.dbo.IDX  que es numérico; "EXT" represnta la extensión por ejemplo "DOC", "PDF"  o la que sea, y "bin" es constante y significa que está comprido. Ejemplos son "345123.PDF.bin", "24512.DOC.bin"
3. Al final genera el fichero con el mismo nombre al que se le ha añadido la extensiónb correspondiente, por ejemplo en los dos casos anteriores generaría "345123.PDF.bin.PDF", "24512.DOC.bin.DOC"

#!/usr/bin/env python3
"""
decode_he3.py
--------------
Codificador/decodificador puro-Python del formato "HE3" usado por
compress.dll (clase clsHuffman, de David Midkiff). Reimplementa
exactamente el algoritmo Huffman original (encontrado en su código
fuente VB6), sin depender de Windows, Wine ni la DLL original.

Verificado byte a byte contra ficheros reales: tanto DecodeFile como
EncodeFile producen un resultado IDÉNTICO al de la DLL original
(incluido el checksum interno del propio formato).

USO:
  Descomprimir un fichero:
      python3 decode_he3.py decode origen.bin destino.pdf

  Comprimir un fichero:
      python3 decode_he3.py encode origen.pdf destino.bin

  Toda una carpeta (descomprimir todos los *.bin):
      python3 decode_he3.py decode --batch carpeta_origen carpeta_destino

  Toda una carpeta (comprimir):
      python3 decode_he3.py encode --batch carpeta_origen carpeta_destino --ext .pdf --suffix .aytos

FORMATO DEL FICHERO (para referencia):
  offset 0-3   : magia "HE3" + CR (0x48 0x45 0x33 0x0D)
                 (si el 3er byte es '0'=0x30, el fichero no se llegó a
                  comprimir y el resto son los datos originales tal cual)
  offset 4     : checksum (XOR de todos los bytes originales)
  offset 5-8   : Long (LE32) = tamaño del fichero original
  offset 9-10  : Integer (LE16) = número de símbolos (valores de byte)
                 distintos usados en el fichero
  luego, por cada símbolo (en orden ascendente de valor de byte):
      1 byte  = valor del símbolo (0-255)
      1 byte  = longitud en bits de su código Huffman
  luego, en el mismo orden ascendente de símbolo:
      los propios patrones de bits de cada código, empaquetados LSB-first,
      concatenados y sin relleno entre símbolos (solo relleno al final,
      hasta completar el byte)
  a partir de ahí, alineado a byte: los datos reales comprimidos,
  empaquetados también LSB-first, decodificables recorriendo el árbol
  binario reconstruido (bit 0 = rama izquierda, bit 1 = rama derecha)
  hasta producir tantos bytes como indica el tamaño original.
"""

import argparse
import os
import struct
import sys


class HE3Error(Exception):
    pass


class _BitReader:
    """Lee bits LSB-first de un buffer, empezando en un byte concreto."""
    __slots__ = ('data', 'pos', 'bitpos')

    def __init__(self, data: bytes, start_byte: int = 0):
        self.data = data
        self.pos = start_byte
        self.bitpos = 0

    def read_bit(self) -> int:
        byte = self.data[self.pos]
        bit = (byte >> self.bitpos) & 1
        self.bitpos += 1
        if self.bitpos == 8:
            self.bitpos = 0
            self.pos += 1
        return bit

    def align_byte(self) -> None:
        if self.bitpos != 0:
            self.bitpos = 0
            self.pos += 1


class _Node:
    __slots__ = ('left', 'right', 'value')

    def __init__(self):
        self.left = None
        self.right = None
        self.value = -1


def decode_he3(data: bytes) -> bytes:
    """Decodifica el contenido binario (formato HE3) y devuelve los bytes
    originales. Lanza HE3Error si la cabecera o el checksum no son validos."""

    if len(data) < 4 or data[0] != 72 or data[1] != 69 or data[3] != 13:
        raise HE3Error("Cabecera invalida: no es un fichero HE0/HE3")

    if data[2] == 48:  # '0' -> el fichero se guardo sin comprimir
        return data[4:]

    if data[2] != 51:  # '3'
        raise HE3Error(f"Version de formato no soportada (byte2={data[2]})")

    checksum = data[4]
    orig_size = struct.unpack('<I', data[5:9])[0]
    if orig_size == 0:
        return b''

    count = struct.unpack('<H', data[9:11])[0]

    pos = 11
    symbols = []
    lengths = {}
    for _ in range(count):
        sym = data[pos]; pos += 1
        length = data[pos]; pos += 1
        symbols.append(sym)
        lengths[sym] = length

    # Leer los patrones de bits de cada codigo (orden raiz->hoja)
    br = _BitReader(data, pos)
    codes = {}
    for sym in symbols:
        bits = [br.read_bit() for _ in range(lengths[sym])]
        codes[sym] = bits
    br.align_byte()

    # Reconstruir el arbol binario: 0 = izquierda, 1 = derecha
    root = _Node()
    for sym, bits in codes.items():
        node = root
        for b in bits:
            if b == 0:
                if node.left is None:
                    node.left = _Node()
                node = node.left
            else:
                if node.right is None:
                    node.right = _Node()
                node = node.right
        node.value = sym

    # Decodificar el resto del fichero recorriendo el arbol
    out = bytearray()
    node = root
    for byte in data[br.pos:]:
        for bitpos in range(8):
            bit = (byte >> bitpos) & 1
            node = node.right if bit else node.left
            if node is None:
                raise HE3Error("Arbol Huffman corrupto durante la decodificacion")
            if node.value > -1:
                out.append(node.value)
                if len(out) == orig_size:
                    chk = 0
                    for b in out:
                        chk ^= b
                    if chk != checksum:
                        raise HE3Error(
                            f"Checksum no coincide (esperado {checksum}, "
                            f"calculado {chk}) - el fichero podria estar corrupto"
                        )
                    return bytes(out)
                node = root
    raise HE3Error("Fin del fichero antes de completar la decodificacion")


def encode_he3(data: bytes) -> bytes:
    """Comprime bytes al formato HE3. Devuelve el mismo resultado, byte a
    byte, que produciria clsHuffman.EncodeFile()."""
    byte_len = len(data)

    if byte_len == 0:
        return bytes([72, 69, 48, 13])  # "HE0" + CR, fichero vacio

    counts = [0] * 256
    for b in data:
        counts[b] += 1

    class _ENode:
        __slots__ = ('weight', 'left', 'right', 'parent', 'value')

        def __init__(self, weight, value=-1, left=-1, right=-1):
            self.weight = weight
            self.value = value
            self.left = left
            self.right = right
            self.parent = -1

    nodes = [_ENode(counts[i], value=i) for i in range(256) if counts[i] > 0]
    nodes_count = len(nodes)
    nodes.extend([None] * (nodes_count - 1))  # espacio para nodos internos

    n = nodes_count
    while n > 1:
        lnode1 = lnode2 = -1
        lweight1 = lweight2 = 0
        # Replica exacta del algoritmo O(n^2) de "dos mas pequenos" del
        # codigo VB original, incluido su criterio de desempate.
        for i, node in enumerate(nodes):
            if node is None or node.parent != -1:
                continue
            if lnode1 == -1:
                lweight1 = node.weight; lnode1 = i
            elif lnode2 == -1:
                lweight2 = node.weight; lnode2 = i
            elif node.weight < lweight1:
                if node.weight < lweight2:
                    if lweight1 < lweight2:
                        lweight2 = node.weight; lnode2 = i
                    else:
                        lweight1 = node.weight; lnode1 = i
                else:
                    lweight1 = node.weight; lnode1 = i
            elif node.weight < lweight2:
                lweight2 = node.weight; lnode2 = i

        new_index = next(i for i, node in enumerate(nodes) if node is None)
        nodes[new_index] = _ENode(lweight1 + lweight2, left=lnode1, right=lnode2)
        nodes[lnode1].parent = new_index
        nodes[lnode2].parent = new_index
        n -= 1

    root_index = next(i for i, node in enumerate(nodes) if node is not None and node.parent == -1)

    codes = {}

    def walk(idx, bits):
        node = nodes[idx]
        if node.value > -1:
            codes[node.value] = bits
            return
        if node.left != -1:
            walk(node.left, bits + [0])
        if node.right != -1:
            walk(node.right, bits + [1])

    walk(root_index, [])
    used_symbols = sorted(codes.keys())

    total_bits = sum(len(codes[i]) * counts[i] for i in used_symbols)
    l_length = (total_bits + 7) // 8

    if l_length == 0 or l_length > byte_len:
        return bytes([72, 69, 48, 13]) + data  # fallback: sin comprimir

    checksum = 0
    for b in data:
        checksum ^= b

    out = bytearray()
    out += bytes([72, 69, 51, 13])
    out.append(checksum)
    out += struct.pack('<I', byte_len)
    out += struct.pack('<H', len(used_symbols))
    for sym in used_symbols:
        out.append(sym)
        out.append(len(codes[sym]))

    bitbuf = 0
    bitcount = 0

    def push_bit(bit):
        nonlocal bitbuf, bitcount
        if bit:
            bitbuf |= (1 << bitcount)
        bitcount += 1
        if bitcount == 8:
            out.append(bitbuf)
            bitbuf = 0
            bitcount = 0

    for sym in used_symbols:
        for bit in codes[sym]:
            push_bit(bit)
    if bitcount > 0:
        out.append(bitbuf)
        bitbuf = 0
        bitcount = 0

    for b in data:
        for bit in codes[b]:
            push_bit(bit)
    if bitcount > 0:
        out.append(bitbuf)

    return bytes(out)


def process_file(src: str, dst: str, mode: str) -> None:
    with open(src, 'rb') as f:
        data = f.read()
    result = decode_he3(data) if mode == 'decode' else encode_he3(data)
    out_dir = os.path.dirname(os.path.abspath(dst))
    if out_dir:
        os.makedirs(out_dir, exist_ok=True)
    with open(dst, 'wb') as f:
        f.write(result)


def main_edu() -> None:
    files=[
        "/home/user/kk/DOCUMENTS_AYTOS/238476.PDF.bin",
        "/home/user/kk/DOCUMENTS_AYTOS/233831.DOC.bin",
    ]
    fail=0
    ok=0
    for fname in files:
        parts=fname.split(".")
        f_out=fname + "." + parts[1]
        try:
            process_file(fname, f_out, "decode")
            print(f"[OK]    {fname} -> {os.path.basename(f_out)}")
            ok += 1
        except Exception as exc:
            print(f"[ERROR] {fname}: {exc}")
            fail += 1
    print(f"\nHecho. {ok} correctos, {fail} con error.")
    

if __name__ == "__main__":
    main_edu()

2. Descargar los ficheros desde BD a disco

Veamos ahora el módulo completo para descargar los ficheros descomprimidos o descifrados a disco

Aquí ya no escribimos el fichero encriptado/comprimido a disco sinó que escribimos el fichero drirectamente a disco con su nombre y extensión correspondiente.
En mi caso he guardado el fichero anterior en la misma carpeta y lo he llamado xmhuffman.py Por tanto la tercera línea se debe de ajustar al nombre que se le haya dado a dicho fichero. En la función main hay que darle los parámetros de acceso a la BD y también la rudta donde guardar los ficheros.

from sqlalchemy import create_engine, text
from urllib.parse import quote_plus
from xmhuffman import decode_he3


# --------------------------------------------------------------------------
# Utilidades de configuración
# --------------------------------------------------------------------------

def _connection_string(c: dict) -> str:
    usuario = quote_plus(str(c["dbUser"]))
    password = quote_plus(str(c["dbPwd"]))
    host = f"{c['dbHost']}:{c['dbPort']}" if c["dbPort"] else c["dbHost"]

    url = f"{c['dbType']}://{usuario}:{password}@{host}/{c['dbName']}"

    params = {}
    if c["dbDriver"]:
        params["driver"] = c["dbDriver"]
    if c["dbTrustServerCertificate"]:
        params["TrustServerCertificate"] = c["dbTrustServerCertificate"]
    if params:
        url += "?" + "&".join(f"{k}={quote_plus(str(v))}" for k, v in params.items())
    return url


def _create_sync_engine(conf: dict):
    """Engine síncrono, para SQL Server (pyodbc)."""
    return create_engine(_connection_string(conf))





def copy_all_blobs2files(
    query_ms: str,
    conf_ms: dict,
    ruta_salida: str,
):
    engine_ms = _create_sync_engine(conf_ms)
    archivos_creados = 0
    
    try:
        with engine_ms.connect() as conn:
            # Usamos fetchall() para traer todos los registros de la consulta
            resultados = conn.execute(text(query_ms)).fetchall()
            
            if not resultados:
                raise ValueError(f"No se encontraron filas para la consulta: {query_ms}")

            i=0
            for fila in resultados:
                if i==2000: 
                    break
                # fila[0] debe ser el ID (o nombre del archivo) y fila[1] el BLOB
                id_registro = fila[0]
                extension = fila[1]
                contenido = fila[2]
                nombre_archivo=f"{id_registro}.{extension}"
                print(nombre_archivo)
                
                # Omitimos registros si el BLOB está vacío
                if contenido is None:
                    print(f"Advertencia: El campo BLOB está vacío para el ID {id_registro}. Saltando...")
                    continue
                
                # Construimos la ruta completa del archivo
                nombre_archivo = f"{ruta_salida}{nombre_archivo}"
                
                result = decode_he3(contenido) 
                with open(nombre_archivo, 'wb') as f:
                    f.write(result)                
                archivos_creados += 1
                i+=1
                
            return archivos_creados
            
    finally:
        engine_ms.dispose()


if __name__ == "__main__":
    conf_ms = {
        "dbType": "mssql+pyodbc",
        "dbUser": "myuser",
        "dbPwd": "mypassword",
        "dbHost": "192.168.xxx.xxx",
        "dbPort": 1433,
        "dbName": "BIB_AY",
        "dbDriver": "ODBC Driver 18 for SQL Server",
        "dbTrustServerCertificate": "yes",
    }

    
    query_ms_all = """
        SELECT B.IDX_ID, B.IDX_EXT, ISNULL(d1.DOC_CNT,d2.ORI_CNT) as doc_blob,B.IDX_ALI, d1.DOC_TAM, d2.ORI_TAM,
            10000*YEAR(B.IDX_FCH)+100*MONTH(B.IDX_FCH)+DAY(B.IDX_FCH) as FOLDER
          FROM BIB_AY.dbo.IDX B
          LEFT JOIN BIB_AY.dbo.DOC d1 ON d1.DOC_ID=B.IDX_ID
          LEFT JOIN BIB_AY.dbo.DOC_ORI d2 ON d2.ORI_ID=B.IDX_ID
          WHERE d1.DOC_TAM>0 or d2.ORI_TAM>0;
        """

    
    resultado =  copy_all_blobs2files(
        query_ms=query_ms_all,
        conf_ms=conf_ms,
        ruta_salida="/home/user/kk/DOCUMENTS_AYTOS/",
    )