feat: interop enc:v1 — decifra e re-cifra o armazenamento do OmniRoute

Dockerfile

@@ -1,38 +1,32 @@
-# ==============================================================================
-# OminiRTKSync: OmniRoute Universal Token & Connection Sync
-# Imagem oficial baseada em Python 3.14 Alpine
-# ==============================================================================
+# OminiRTKSync — guardião de conexões do OmniRoute (fonte vendored via git
+# subtree de pathbit/OminiRTkSync). A imagem base é pinada por digest como
+# qualquer outra dependência externa (contrato SEC-01).
+FROM python:3.14-alpine@sha256:c6ead215bfd31f1e433d968853b7a769989117115b728874824e6c0a27cb96fc
 
-FROM python:3.14-alpine
-
-LABEL org.opencontainers.image.title="OminiRTKSync"
-LABEL org.opencontainers.image.description="OminiRoute Universal Token & Connection Synchronizer"
-LABEL org.opencontainers.image.authors="Eliel Sousa <eliel@pathbit.co>"
-LABEL org.opencontainers.image.source="https://github.com/pathbit/OminiRTkSync"
+LABEL org.opencontainers.image.title="TalqueeAI OmniRTKSync"
+LABEL org.opencontainers.image.description="OminiRTKSync com base pinada por digest e painel restrito ao loopback"
+LABEL org.opencontainers.image.version="1.0.0-talquee.1"
 
 WORKDIR /app
 
-# Criação obrigatória e isolada do Virtual Environment
 RUN python3 -m venv /opt/venv
-ENV PATH="/opt/venv/bin:$PATH"
-ENV VIRTUAL_ENV="/opt/venv"
-
-ENV PYTHONUNBUFFERED=1
-ENV PYTHONPATH=/app/src
-ENV DB_PATH=/app/data/storage.sqlite
-ENV OMNIROUTE_URL=http://127.0.0.1:20128
-ENV SYNC_INTERVAL=300
-ENV REFRESH_MARGIN=900
-ENV WEB_PORT=9090
-ENV WEB_HOST=0.0.0.0
-ENV ENABLE_WEB_DASHBOARD=1
-
-COPY src/ /app/src/
-COPY pyproject.toml /app/
-
-# Instalação do pacote dentro do virtual environment
-RUN pip install --no-cache-dir --upgrade pip && \
-    pip install --no-cache-dir -e .
+ENV PATH="/opt/venv/bin:$PATH" \
+    VIRTUAL_ENV="/opt/venv" \
+    PYTHONUNBUFFERED=1 \
+    PYTHONPATH=/app/src \
+    DB_PATH=/app/data/storage.sqlite \
+    OMNIROUTE_URL=http://omniroute:20128 \
+    SYNC_INTERVAL=300 \
+    REFRESH_MARGIN=900 \
+    WEB_PORT=9090 \
+    WEB_HOST=127.0.0.1 \
+    ENABLE_WEB_DASHBOARD=1
+
+COPY apps/ominirtksync/src/ /app/src/
+COPY apps/ominirtksync/pyproject.toml /app/
+
+RUN pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir -e .
 
 EXPOSE 9090
 

pyproject.toml

@@ -12,6 +12,10 @@ authors = [
 ]
 license = { text = "MIT" }
 requires-python = ">=3.10"
+dependencies = [
+    # AES-256-GCM para decifrar/cifrar o armazenamento enc:v1 do OmniRoute.
+    "pycryptodome==3.23.0",
+]
 keywords = ["omniroute", "oauth", "token-sync", "llm", "antigravity", "claude-code"]
 classifiers = [
     "Programming Language :: Python :: 3",

src/omini_rtksync/database.py

@@ -7,6 +7,8 @@
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
 
+from .enc_v1 import decrypt_if_needed, encrypt
+
 
 def get_db_connection(db_path: str) -> sqlite3.Connection:
     if not os.path.exists(db_path):
@@ -123,6 +125,13 @@ def get_all_connections(db_path: str) -> List[Dict[str, Any]]:
                 extra.get("providerSpecificData")
             )
 
+            # Credenciais em repouso são cifradas pelo OmniRoute (enc:v1). A
+            # leitura decifra com a chave de ambiente; o que não decifra volta
+            # intacto, e nenhum caminho joga o material em log.
+            access_token = decrypt_if_needed(access_token) if access_token else access_token
+            refresh_token = decrypt_if_needed(refresh_token) if refresh_token else refresh_token
+            api_key = decrypt_if_needed(api_key) if api_key else api_key
+
             result.append({
                 "id": str(item["id"]),
                 "provider": provider,
@@ -209,6 +218,10 @@ def update_connection(
     conn = get_db_connection(db_path)
     tbl = detect_connection_table(conn)
     now_iso = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+    # A escrita volta cifrada no mesmo formato do OmniRoute (enc:v1) quando a
+    # chave está presente; sem ela, passthrough em claro, como o gateway faz.
+    access_token = encrypt(access_token)
+    refresh_token = encrypt(refresh_token)
     try:
         cursor = conn.cursor()
         cursor.execute(f"PRAGMA table_info({tbl})")

src/omini_rtksync/enc_v1.py

@@ -0,0 +1,93 @@
+"""Decifra e cifra campos `enc:v1:` do armazenamento do OmniRoute.
+
+O OmniRoute cifra credenciais em repouso com AES-256-GCM. Formato (medido no
+bundle do servidor):
+
+    enc:v1:<iv_hex>:<ciphertext_hex>:<auth_tag_hex>
+
+Derivação da chave (linha exata do bundle: scryptSync(key, contexto, 32)):
+
+    chave32 = scrypt(senha=STORAGE_ENCRYPTION_KEY,
+                     sal="omniroute-field-encryption-v1",
+                     n=16384, r=8, p=1, dklen=32)
+
+O sal é o CONTEXTO estático acima — o sha256 da chave só aparece no caminho
+legado de migração, não no formato vigente. A chave chega por variável de
+ambiente e nunca vai a log: falhas devolvem None ou o valor original, sem
+detalhe do material.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import secrets as _secrets
+
+from Crypto.Cipher import AES
+
+PREFIX = "enc:v1:"
+CONTEXT_SALT = b"omniroute-field-encryption-v1"
+
+
+def _derive_key(secret: str) -> bytes:
+    return hashlib.scrypt(
+        secret.encode("utf-8"),
+        salt=CONTEXT_SALT,
+        n=16384,
+        r=8,
+        p=1,
+        dklen=32,
+    )
+
+
+def is_encrypted(value: object) -> bool:
+    return isinstance(value, str) and value.startswith(PREFIX)
+
+
+def decrypt(value: str, *, secret: str | None = None) -> str | None:
+    """Decifra um valor `enc:v1:`; None quando não decifra (nunca levanta).
+
+    Valores fora do formato passam intactos (passthrough), como o próprio
+    OmniRoute faz quando a chave está ausente.
+    """
+
+    if not is_encrypted(value):
+        return value
+    secret = secret if secret is not None else os.environ.get("STORAGE_ENCRYPTION_KEY", "")
+    if not secret:
+        return None
+    parts = value[len(PREFIX) :].split(":")
+    if len(parts) != 3:
+        return None
+    iv_hex, ciphertext_hex, tag_hex = parts
+    try:
+        iv = bytes.fromhex(iv_hex)
+        ciphertext = bytes.fromhex(ciphertext_hex)
+        tag = bytes.fromhex(tag_hex)
+        cipher = AES.new(_derive_key(secret), AES.MODE_GCM, nonce=iv)
+        plaintext = cipher.decrypt(ciphertext)
+        cipher.verify(tag)
+        return plaintext.decode("utf-8")
+    except (ValueError, TypeError, UnicodeDecodeError):
+        return None
+
+
+def decrypt_if_needed(value: str) -> str:
+    """Atalho: decifra quando cifrado; devolve o original quando ilegível."""
+
+    result = decrypt(value)
+    return result if result is not None else value
+
+
+def encrypt(value: str, *, secret: str | None = None) -> str:
+    """Cifra no formato `enc:v1:`; sem chave, devolve o original (passthrough)."""
+
+    if not value or is_encrypted(value):
+        return value
+    secret = secret if secret is not None else os.environ.get("STORAGE_ENCRYPTION_KEY", "")
+    if not secret:
+        return value
+    iv = _secrets.token_bytes(16)
+    cipher = AES.new(_derive_key(secret), AES.MODE_GCM, nonce=iv)
+    ciphertext, tag = cipher.encrypt_and_digest(value.encode("utf-8"))
+    return f"{PREFIX}{iv.hex()}:{ciphertext.hex()}:{tag.hex()}"

tests/test_enc_v1.py

@@ -0,0 +1,57 @@
+"""enc:v1 — interop de cifra com o armazenamento do OmniRoute.
+
+O vetor INTEROP foi gerado com o scryptSync + AES-256-GCM do Node.js (os
+mesmos parâmetros do bundle do OmniRoute) contra a chave de teste — a prova
+de que o Python lê exatamente o que o gateway escreve.
+"""
+
+from __future__ import annotations
+
+import os
+import unittest
+
+from omini_rtksync.enc_v1 import decrypt, decrypt_if_needed, encrypt, is_encrypted
+
+CHAVE = "teste-chave-de-campo-32bytes"
+INTEROP_NODE = "enc:v1:b321265c13c4321569e5def33f2e5e46:0101a8fc1b2ae325092ffed20fdfcd0920abb6e9:573d1423166eee97e8eab80aa4851b5a"
+SEGREDO = "sk-teste-segredo-123"
+
+
+class EncV1InteropTest(unittest.TestCase):
+    def setUp(self) -> None:
+        os.environ["STORAGE_ENCRYPTION_KEY"] = CHAVE
+
+    def tearDown(self) -> None:
+        os.environ.pop("STORAGE_ENCRYPTION_KEY", None)
+
+    def test_decifra_valor_produzido_pelo_node(self) -> None:
+        self.assertEqual(decrypt(INTEROP_NODE), SEGREDO)
+
+    def test_round_trip_cifra_e_decifra(self) -> None:
+        cifrado = encrypt(SEGREDO)
+        self.assertTrue(is_encrypted(cifrado))
+        self.assertEqual(decrypt(cifrado), SEGREDO)
+
+    def test_cifras_sao_unicas_por_iv_aleatorio(self) -> None:
+        self.assertNotEqual(encrypt("mesmo"), encrypt("mesmo"))
+
+    def test_sem_chave_passthrough_e_none(self) -> None:
+        os.environ.pop("STORAGE_ENCRYPTION_KEY", None)
+        self.assertIsNone(decrypt(INTEROP_NODE))
+        self.assertEqual(encrypt("claro"), "claro")
+
+    def test_valor_em_claro_passa_intacto(self) -> None:
+        self.assertEqual(decrypt("texto-claro"), "texto-claro")
+        self.assertEqual(decrypt_if_needed("texto-claro"), "texto-claro")
+
+    def test_malformado_nao_levanta(self) -> None:
+        self.assertIsNone(decrypt("enc:v1:aa:bb"))
+        self.assertEqual(decrypt_if_needed("enc:v1:aa:bb"), "enc:v1:aa:bb")
+
+    def test_tag_adulterada_recusa(self) -> None:
+        adulterado = INTEROP_NODE[:-1] + ("0" if INTEROP_NODE[-1] != "0" else "1")
+        self.assertIsNone(decrypt(adulterado))
+
+
+if __name__ == "__main__":
+    unittest.main()