from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
import os
# 1. Definir as ferramentas (@tools) que a IA pode invocar de forma autónoma
@tool
def executar_comando_seguro(comando: str) -> str:
"""Executa um comando de diagnóstico ou ferramenta de análise no terminal e devolve o output."""
# Exemplo: Se fosses integrar ferramentas de segurança/hacking ético (ex: nmap, ping, traceroute)
# Nota: Executa sempre comandos num ambiente isolado (sandbox)!
import subprocess
try:
resultado = subprocess.run(comando, shell=True, capture_output=True, text=True, timeout=10)
return resultado.stdout if resultado.stdout else resultado.stderr
except Exception as e:
return f"Erro ao executar: {str(e)}"
# Agrupar as ferramentas numa lista
lista_de_ferramentas = [executar_comando_seguro]
# 2. Configurar o Modelo de IA e o Prompt Base com suporte a histórico/scratchpad
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "És um agente de IA especialista em automação e cibersegurança. "
"Tens permissão para usar as ferramentas fornecidas para analisar e resolver o problema."),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"), # Guarda o raciocínio interno do agente
])
# 3. Construir e interligar o Agente Autónomo
agente = create_openai_tools_agent(llm, lista_de_ferramentas, prompt)
executor_agente = AgentExecutor(agent=agente, tools=lista_de_ferramentas, verbose=True)
# 4. Execução da automação
# O agente decidirá SOZINHO se precisa de invocar a ferramenta e que parâmetros usar
resposta = executor_agente.invoke({"input": "Verifica se o servidor local na porta 8080 está ativo usando um ping.", "chat_history": []})
print(resposta["output"])
@tool
def ping(host: str) -> str:
"""Executa ping ICMP ao host."""
import subprocess
comando = f"ping -c 4 {host}"
resultado = subprocess.run(comando, shell=True, capture_output=True, text=True)
return resultado.stdout + resultado.stderr
@tool
def testar_porta(host: str, porta: int) -> str:
"""Verifica se uma porta TCP está aberta usando netcat."""
import subprocess
comando = f"nc -zv {host} {porta}"
resultado = subprocess.run(comando, shell=True, capture_output=True, text=True)
return resultado.stdout + resultado.stderr
@tool
def http_head(url: str) -> str:
"""Faz um pedido HTTP HEAD para verificar disponibilidade."""
import subprocess
comando = f"curl -I {url}"
resultado = subprocess.run(comando, shell=True, capture_output=True, text=True)
return resultado.stdout + resultado.stderr
@tool
def nmap_scan(host: str, porta: int) -> str:
"""Executa um scan nmap à porta especificada."""
import subprocess
comando = f"nmap -p {porta} {host}"
resultado = subprocess.run(comando, shell=True, capture_output=True, text=True)
return resultado.stdout + resultado.stderr
@tool
def nmap_scan(host: str, porta: int) -> str:
"""Executa um scan nmap à porta especificada."""
import subprocess
comando = f"nmap -p {porta} {host}"
resultado = subprocess.run(comando, shell=True, capture_output=True, text=True)
return resultado.stdout + resultado.stderr
@tool
def docker_exec(cmd: str) -> str:
import subprocess
return subprocess.run(f"docker exec {cmd}", shell=True, capture_output=True, text=True).stdout
@tool
def docker_export(container: str) -> str:
import subprocess
return subprocess.run(f"docker export {container} -o /tmp/{container}.tar",
shell=True, capture_output=True, text=True).stdout
from fastapi import FastAPI
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
import subprocess
app = FastAPI()
# ---------------------------
# FERRAMENTAS DOCKER REAIS
# ---------------------------
@tool
def docker_ps(_: str) -> str:
"""Lista containers ativos."""
return subprocess.run("docker ps", shell=True, capture_output=True, text=True).stdout
@tool
def docker_inspect(container: str) -> str:
"""Inspeciona container."""
return subprocess.run(f"docker inspect {container}", shell=True, capture_output=True, text=True).stdout
@tool
def docker_export(container: str) -> str:
"""Exporta filesystem do container."""
out = f"/tmp/{container}.tar"
subprocess.run(f"docker export {container} -o {out}", shell=True)
return f"Exportado para {out}"
@tool
def docker_save(image: str) -> str:
"""Exporta imagem Docker."""
out = f"/tmp/{image}.tar"
subprocess.run(f"docker save {image} -o {out}", shell=True)
return f"Imagem guardada em {out}"
@tool
def docker_layers(image: str) -> str:
"""Mostra layers da imagem."""
return subprocess.run(f"docker history {image}", shell=True, capture_output=True, text=True).stdout
@tool
def docker_exec(cmd: str) -> str:
"""Executa comando dentro do container."""
return subprocess.run(f"docker exec {cmd}", shell=True, capture_output=True, text=True).stdout
# Lista de ferramentas
tools = [
docker_ps,
docker_inspect,
docker_export,
docker_save,
docker_layers,
docker_exec
]
# ---------------------------
# AGENTE LANGCHAIN
# ---------------------------
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system",
"És um agente de cibersegurança especializado em análise Docker. "
"Decide autonomamente que ferramenta usar. "
"Se o utilizador pedir para abrir, desmontar ou analisar um Docker, "
"usa docker_ps, docker_inspect, docker_export, docker_save, docker_layers ou docker_exec."),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# ---------------------------
# ENDPOINT PÚBLICO
# ---------------------------
@app.post("/docker")
def analisar_docker(input: str):
resposta = executor.invoke({"input": input, "chat_history": []})
return {"output": resposta["output"]}
<!DOCTYPE html>
<html>
<head>
<title>Docker Analyzer</title>
<style>
body { font-family: Arial; background: #111; color: #0f0; padding: 20px; }
textarea { width: 100%; height: 120px; background: #000; color: #0f0; }
button { padding: 10px 20px; background: #0f0; border: none; cursor: pointer; }
pre { background: #000; color: #0f0; padding: 20px; }
</style>
</head>
<body>
<h1>Docker Analyzer</h1>
<textarea id="input" placeholder="Escreve o comando: ex. 'Abrir container nginx'"></textarea>
<br><br>
<button onclick="send()">Executar</button>
<h2>Output:</h2>
<pre id="output"></pre>
<script>
async function send() {
const input = document.getElementById("input").value;
const res = await fetch("/docker", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ input })
});
const data = await res.json();
document.getElementById("output").innerText = data.output;
}
</script>
</body>
</html>
Abre o container nginx e mostra as layers da imagem.
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
e3f1c2... nginx ...
Layer 1: base filesystem
Layer 2: apt updates
Layer 3: config copy
Layer 4: entrypoint
No comments:
Post a Comment