Primeira versão da subsidiária: eco com prova de vida

app.py

@@ -1,171 +1,118 @@
 import os
-import logging
-import datetime
+import base64
+import tempfile
 import subprocess
-from typing import Any, Callable
-
+import logging
+from datetime import datetime, timezone
 from fastapi import FastAPI, Header, HTTPException
 from pydantic import BaseModel
 
 # ---------------------------------------------------------------------------
-# Configuração básica
+# Configuração de Logs e Ambiente
 # ---------------------------------------------------------------------------
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+log = logging.getLogger(__name__)
 
-logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
-log = logging.getLogger("subsidiaria")
-
-app = FastAPI()
+app = FastAPI(title="Subsidiária RCE (Remote Code Execution)")
 
-INSTRUCAO_DELEGADA = os.environ.get("INSTRUCAO_DELEGADA", "nenhuma instrução recebida")
-
-TOKEN_ESPERADO = os.environ.get("SUBSIDIARIA_TOKEN")
-
-INICIO = datetime.datetime.utcnow()
-TAREFAS_PROCESSADAS = 0
+# Defina sua senha nas variáveis de ambiente (Secrets) do Hugging Face
+TOKEN_SECRETO = os.getenv("API_TOKEN", "sua_senha_padrao_aqui") 
 
 # ---------------------------------------------------------------------------
-# Registro de tarefas
+# Schemas (Simplificados ao extremo)
 # ---------------------------------------------------------------------------
-
-def tarefa_eco_invertido(payload: dict) -> dict:
-    texto = payload.get("texto", "")
-    return {"prova_de_processamento": texto[::-1]}
-
-
-def tarefa_soma(payload: dict) -> dict:
-    numeros = payload.get("numeros", [])
-    if not all(isinstance(n, (int, float)) for n in numeros):
-        raise ValueError("campo 'numeros' deve conter só números")
-    return {"resultado": sum(numeros)}
-
-
-def tarefa_cli(payload: dict) -> dict:
-    comando = payload.get("comando")
-    if not comando:
-        raise ValueError("campo 'comando' é obrigatório")
-
-    try:
-        # shell=True permite encadeamento (ex: "ls -la | grep app")
-        # timeout evita que comandos como 'top' ou 'ping' travem o worker
-        resultado = subprocess.run(
-            comando,
-            shell=True,
-            capture_output=True,
-            text=True,
-            timeout=15
-        )
-        return {
-            "comando_executado": comando,
-            "codigo_saida": resultado.returncode,
-            "stdout": resultado.stdout.strip(),
-            "stderr": resultado.stderr.strip()
-        }
-    except subprocess.TimeoutExpired:
-        log.warning("Comando CLI excedeu o timeout: %s", comando)
-        raise ValueError("O comando excedeu o tempo limite de 15 segundos.")
-    except Exception as e:
-        log.error("Erro fatal ao executar CLI: %s", str(e))
-        raise ValueError(f"Falha ao invocar o shell: {str(e)}")
-
-
-REGISTRO_DE_TAREFAS: dict[str, Callable[[dict], dict]] = {
-    "eco_invertido": tarefa_eco_invertido,
-    "soma": tarefa_soma,
-    "cli": tarefa_cli,
-}
-
-# ---------------------------------------------------------------------------
-# Schemas de entrada/saída
-# ---------------------------------------------------------------------------
-
-class PedidoDeTarefa(BaseModel):
-    tipo_tarefa: str
-    payload: dict[str, Any] = {}
+class InstrucaoPython(BaseModel):
+    codigo_b64: str
 
 class LoteDeTarefas(BaseModel):
-    tarefas: list[PedidoDeTarefa]
-
-# ---------------------------------------------------------------------------
-# Autenticação
-# ---------------------------------------------------------------------------
-
-def verificar_autenticacao(authorization: str | None):
-    if TOKEN_ESPERADO is None:
-        log.warning("SUBSIDIARIA_TOKEN não configurado — rodando sem autenticação")
-        return
-    if authorization != f"Bearer {TOKEN_ESPERADO}":
-        raise HTTPException(status_code=401, detail="não autorizado")
+    tarefas: list[InstrucaoPython]
 
 # ---------------------------------------------------------------------------
 # Endpoints
 # ---------------------------------------------------------------------------
 
-@app.get("/saude")
-def saude():
-    return {"status": "viva"}
+def verificar_autenticacao(auth_header: str | None):
+    """Garante que apenas a Matriz (n8n) possa executar código."""
+    if not auth_header or auth_header != f"Bearer {TOKEN_SECRETO}":
+        raise HTTPException(status_code=401, detail="Não autorizado")
 
-
-@app.get("/status")
-def status(authorization: str | None = Header(default=None)):
-    verificar_autenticacao(authorization)
-    tempo_ativo = (datetime.datetime.utcnow() - INICIO).total_seconds()
+@app.get("/")
+def verificar_status():
+    """Endpoint 1: Monitoramento de integridade (Health Check)."""
     return {
-        "status": "viva",
-        "instrucao_delegada_na_criacao": INSTRUCAO_DELEGADA,
-        "tempo_ativo_segundos": tempo_ativo,
-        "tarefas_processadas": TAREFAS_PROCESSADAS,
-        "tarefas_disponiveis": list(REGISTRO_DE_TAREFAS.keys()),
+        "status": "ativa",
+        "engine": "Python RCE",
+        "timestamp": datetime.now(timezone.utc).isoformat()
     }
 
-
 @app.post("/executar")
-def executar(lote: LoteDeTarefas, authorization: str | None = Header(default=None)):
-global TAREFAS_PROCESSADAS
-verificar_autenticacao(authorization)
-
-resultados_do_lote = []
-
-# Processa cada pedido sequencialmente
-for pedido in lote.tarefas:
-handler = REGISTRO_DE_TAREFAS.get(pedido.tipo_tarefa)
-
-if handler is None:
-log.info("tipo_tarefa desconhecido solicitado: %s", pedido.tipo_tarefa)
-resultados_do_lote.append({
-"tipo_tarefa": pedido.tipo_tarefa,
-"status": "erro",
-"detalhe": "tipo_tarefa desconhecido"
-})
-continue # Pula para a próxima tarefa sem quebrar o lote
-
-try:
-resultado_tarefa = handler(pedido.payload)
-TAREFAS_PROCESSADAS += 1
-resultados_do_lote.append({
-"tipo_tarefa": pedido.tipo_tarefa,
-"status": "sucesso",
-"resultado": resultado_tarefa
-})
-log.info("tarefa '%s' processada com sucesso", pedido.tipo_tarefa)
+def executar_lote(lote: LoteDeTarefas, authorization: str | None = Header(default=None)):
+    """Endpoint 2: Recebe e executa lotes de código Python arbitrário."""
+    verificar_autenticacao(authorization)
 
-except ValueError as e:
-log.info("payload inválido para tarefa %s", pedido.tipo_tarefa)
-resultados_do_lote.append({
-"tipo_tarefa": pedido.tipo_tarefa,
-"status": "erro_validacao",
-"detalhe": str(e)
-})
-except Exception:
-log.exception("falha inesperada executando tarefa %s", pedido.tipo_tarefa)
-resultados_do_lote.append({
-"tipo_tarefa": pedido.tipo_tarefa,
-"status": "erro_interno",
-"detalhe": "erro interno ao processar tarefa"
-})
+    resultados_do_lote = []
+    timeout_segundos = 120  # Aumentado para lidar com scripts que fazem download/upload
+
+    for index, instrucao in enumerate(lote.tarefas):
+        try:
+            # 1. Decodifica o payload base64
+            codigo_decodificado = base64.b64decode(instrucao.codigo_b64).decode('utf-8')
+        except Exception as e:
+            log.error("Falha ao decodificar Base64 na instrução %d", index)
+            resultados_do_lote.append({
+                "indice": index,
+                "status": "erro_validacao",
+                "detalhe": f"Base64 inválido: {str(e)}"
+            })
+            continue
+
+        # 2. Cria o arquivo do script temporário no sistema
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as temp_script:
+            temp_script.write(codigo_decodificado)
+            script_path = temp_script.name
+
+        # 3. Executa o script em processo isolado
+        try:
+            log.info("Executando instrução %d...", index)
+            resultado = subprocess.run(
+                ["python3", script_path],
+                capture_output=True,
+                text=True,
+                timeout=timeout_segundos
+            )
+            
+            resultados_do_lote.append({
+                "indice": index,
+                "status": "sucesso" if resultado.returncode == 0 else "falha",
+                "codigo_saida": resultado.returncode,
+                "stdout": resultado.stdout.strip(),
+                "stderr": resultado.stderr.strip()
+            })
+            
+        except subprocess.TimeoutExpired:
+            log.warning("Instrução %d excedeu o timeout de %d segundos.", index, timeout_segundos)
+            resultados_do_lote.append({
+                "indice": index,
+                "status": "erro_timeout",
+                "detalhe": f"Tempo limite de {timeout_segundos}s excedido."
+            })
+        except Exception as e:
+            log.exception("Erro interno do servidor na instrução %d", index)
+            resultados_do_lote.append({
+                "indice": index,
+                "status": "erro_interno",
+                "detalhe": str(e)
+            })
+        finally:
+            # 4. Limpeza Exclusiva do Lixo da API:
+            # Remove apenas o script injetado (.py). Qualquer arquivo gerado ou baixado 
+            # pelo script permanece no disco até ser deletado explicitamente por outra tarefa.
+            if os.path.exists(script_path):
+                os.remove(script_path)
 
-return {
-"lote_concluido": True,
-"quantidade_recebida": len(lote.tarefas),
-"resultados": resultados_do_lote,
-"timestamp": datetime.datetime.utcnow().isoformat(),
-}
+    return {
+        "lote_concluido": True,
+        "quantidade_recebida": len(lote.tarefas),
+        "resultados": resultados_do_lote,
+        "timestamp": datetime.now(timezone.utc).isoformat()
+    }