David
f9384c9376
- Added `INVENTORY_DETAIL_BUDGET` and `RESOURCE_DETAIL_BUDGET` to define layout parameters for inventory and resource detail views. - Implemented `useContextualDetail` hook to manage visibility and focus transitions between inline and overlay modes. - Created `contextual-detail-layout` module to handle budget calculations and state management for contextual details. - Updated documentation to reflect changes in asset and resource handling, ensuring consistent UI behavior across different viewports. - Conducted regression tests to validate functionality and ensure no existing features were broken during the implementation.
382 lines
21 KiB
Python
382 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Prepara datos desechables y verifica un acceso real mediante Playwright MCP."""
|
|
|
|
import argparse
|
|
import base64
|
|
import contextlib
|
|
import hashlib
|
|
import http.cookiejar
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import runpy
|
|
import secrets
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import tomllib
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
ARTIFACTS = ROOT / ".artifacts/playwright"
|
|
# Índice oficial contrastado con el registro; no utiliza el volumen de compose del proyecto.
|
|
DATABASE_IMAGE = "mariadb:11.8@sha256:6422478cb8e159f080fb1d8ccf65101e26fe51385787fde7d16c3b165a331f15"
|
|
|
|
|
|
def private_write(path, text):
|
|
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
|
|
stream.write(text)
|
|
|
|
|
|
def command(args, **options):
|
|
return subprocess.run(args, check=True, timeout=120, capture_output=True, text=True, **options)
|
|
|
|
|
|
def available_port():
|
|
with socket.socket() as listener:
|
|
listener.bind(("127.0.0.1", 0))
|
|
return listener.getsockname()[1]
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def child_process(args, environment, log):
|
|
with log.open("w", encoding="utf-8") as stream:
|
|
process = subprocess.Popen(args, cwd=ROOT, env=environment, stdout=stream,
|
|
stderr=stream, start_new_session=True)
|
|
try:
|
|
yield process
|
|
finally:
|
|
if process.poll() is None:
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
try:
|
|
process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
process.wait(timeout=10)
|
|
|
|
|
|
def wait_for(check, process=None):
|
|
deadline = time.monotonic() + 120
|
|
while time.monotonic() < deadline:
|
|
if process is not None and process.poll() is not None:
|
|
raise RuntimeError("Un servidor temporal terminó antes de estar disponible; revisa sus logs locales.")
|
|
if check():
|
|
return
|
|
# Reintento condicionado a disponibilidad, con plazo; no sustituye una espera de UI.
|
|
time.sleep(0.25)
|
|
raise RuntimeError("Tiempo agotado esperando un servicio temporal.")
|
|
|
|
|
|
class TestApi:
|
|
"""Cliente del HTTP público; nunca fabrica cookies, sesiones ni tokens CSRF."""
|
|
|
|
def __init__(self, base, origin):
|
|
self.base = base
|
|
self.origin = origin
|
|
self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
|
|
|
|
def request(self, path, body=None, expected=200):
|
|
headers = {}
|
|
if body is not None:
|
|
token = self.request("/auth/session")["csrfToken"]
|
|
headers = {"content-type": "application/json", "origin": self.origin, "x-csrf-token": token}
|
|
request = urllib.request.Request(self.base + "/api" + path,
|
|
data=None if body is None else json.dumps(body).encode(),
|
|
headers=headers)
|
|
try:
|
|
with self.opener.open(request, timeout=20) as response:
|
|
if response.status != expected:
|
|
raise RuntimeError("Respuesta HTTP inesperada en " + path)
|
|
return json.load(response)
|
|
except urllib.error.HTTPError as error:
|
|
if error.code == expected:
|
|
return None
|
|
# Los cuerpos pueden contener datos de sesión; no se vuelcan en el diagnóstico.
|
|
raise RuntimeError(f"HTTP {error.code} en {path}") from None
|
|
|
|
def ready(self):
|
|
try:
|
|
self.request("/auth/session")
|
|
return True
|
|
except (OSError, RuntimeError):
|
|
return False
|
|
|
|
|
|
def prepare_resources(api, company_id):
|
|
"""Fixtures documentales del catálogo real, repetibles dentro de la instalación desechable."""
|
|
catalog = api.request("/resource-catalog/" + company_id)["items"]
|
|
definitions = {item["builtinKey"]: item for item in catalog if item.get("builtinKey")}
|
|
existing = {item["values"]["name"]: item for item in api.request("/resources/" + company_id)["resources"]}
|
|
samples = [
|
|
("application", f"AUDIT-APP-{index:02d}", index % 4 != 0,
|
|
{"provider": "Proveedor sintético", "url": "https://example.test/aplicaciones",
|
|
"description": "Aplicación sintética para comparación y paginación."})
|
|
for index in range(1, 13)
|
|
] + [
|
|
("shared_folder", "AUDIT-FOLDER — Documentación compartida de soporte e infraestructura de la compañía", True,
|
|
{"unc": r"\\AUDIT-NAS\documentacion", "server": "AUDIT-NAS",
|
|
"description": "Carpeta documental sintética; sin publicación física ni permisos inventados."}),
|
|
("software_license", "AUDIT-LICENSE — Suite empresarial de pruebas", True,
|
|
{"software_product": "Suite sintética", "software_edition": "Enterprise de pruebas",
|
|
"provider": "Proveedor sintético", "license_model": "Suscripción",
|
|
"license_metric": "Por usuario", "license_quantity": 0, "auto_renew": False,
|
|
"reference": "AUDIT-LIC-01", "expires": "2027-12-31", "renews": "2027-12-01"}),
|
|
("certificate", "AUDIT-CERT — Certificado retirado", False,
|
|
{"holder": "Compañía sintética", "issuer": "Emisor de pruebas", "expires": "2025-01-01"}),
|
|
("windows_domain", "AUDIT-DOMAIN", True,
|
|
{"domain_dns": "audit.example.test", "domain_netbios": "AUDIT"}),
|
|
]
|
|
resources = []
|
|
for kind, name, active, values in samples:
|
|
template = definitions.get("template:" + kind)
|
|
if template is None:
|
|
raise RuntimeError("Falta una plantilla de serie para la fixture " + kind)
|
|
characteristics = {}
|
|
for key, value in values.items():
|
|
field = definitions.get("field:" + key)
|
|
if field is None:
|
|
raise RuntimeError("Falta un campo de serie para la fixture " + key)
|
|
characteristics[field["id"]] = value
|
|
resource = existing.get(name)
|
|
if resource is None:
|
|
resource = api.request("/resources/" + company_id, {"values": {
|
|
"name": name, "templateId": template["id"], "active": active,
|
|
"notes": "Documentación sintética para comprobar desplazamiento y continuidad del contexto. " * 30,
|
|
"characteristics": characteristics,
|
|
}}, expected=201)["resource"]
|
|
existing[name] = resource
|
|
resources.append({"id": resource["id"], "name": name, "kind": kind, "active": active})
|
|
return resources
|
|
|
|
|
|
def prepare_fixture(api):
|
|
if not api.request("/auth/session")["setupRequired"]:
|
|
raise RuntimeError("La instancia desechable no está vacía; se rechaza preparar datos.")
|
|
password = secrets.token_urlsafe(32)
|
|
api.request("/auth/bootstrap", {"name": "Preparación temporal", "email": "setup@audit.example.test",
|
|
"password": secrets.token_urlsafe(32)}, expected=201)
|
|
user = api.request("/users", {"name": "Auditoría visual", "email": "visual@audit.example.test",
|
|
"password": password}, expected=201)["user"]
|
|
if user["isInitialAdmin"]:
|
|
raise RuntimeError("La cuenta de auditoría debe ser ordinaria.")
|
|
company = api.request("/companies", {"name": "Auditoría visual — Datos sintéticos", "active": True}, expected=201)["company"]
|
|
company_id = str(uuid.UUID(company["id"]))
|
|
catalog = api.request("/asset-catalog/" + company_id)["items"]
|
|
template = next((item for item in catalog if item["values"]["kind"] == "template"
|
|
and item["values"]["name"] == "Portátil"), None)
|
|
if template is None:
|
|
raise RuntimeError("El catálogo de serie no incluye la plantilla Portátil.")
|
|
asset = api.request("/assets/" + company_id, {"values": {
|
|
"name": "AUDIT-PORT-01", "templateId": template["id"], "siteId": None,
|
|
"registeredOn": "2026-01-15", "status": "operational", "notes": "Activo sintético para auditorías visuales.",
|
|
"characteristics": {},
|
|
}}, expected=201)["asset"]
|
|
resources = prepare_resources(api, company_id)
|
|
api.request("/auth/logout", {})
|
|
return {"company": company_id, "asset": str(uuid.UUID(asset["id"])), "user": user["id"],
|
|
"email": "visual@audit.example.test", "password": password, "resources": resources}
|
|
|
|
|
|
def tool(client, name, arguments):
|
|
try:
|
|
return client.tool(name, arguments)
|
|
except (RuntimeError, OSError) as error:
|
|
# Las respuestas de rellenado pueden repetir credenciales. No imprimirlas ni guardarlas.
|
|
if name == "browser_run_code_unsafe":
|
|
# Solo categorías de error conocidas; nunca contenido, argumentos ni snapshots.
|
|
category = "timeout" if "Timeout" in str(error) else "ejecución de código"
|
|
raise RuntimeError("Falló " + name + " (" + category + ").") from None
|
|
raise RuntimeError("Falló " + name + "; no se registra su contenido para proteger la sesión.") from None
|
|
|
|
|
|
def result_json(result):
|
|
for part in result.get("content", []):
|
|
if part.get("type") == "text" and "### Result\n" in part["text"]:
|
|
return json.JSONDecoder().raw_decode(part["text"].split("### Result\n", 1)[1].lstrip())[0]
|
|
raise RuntimeError("El MCP no devolvió el resultado JSON esperado.")
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def browser(config, state=None):
|
|
name = "metis-visual-browser-" + uuid.uuid4().hex[:12]
|
|
args = ["docker", "run", "--name=" + name, *config["args"][1:]]
|
|
if state is not None:
|
|
image_position = next(index for index, arg in enumerate(args) if arg.startswith("mcr.microsoft.com/playwright/mcp:"))
|
|
args[image_position:image_position] = ["--user=" + str(os.getuid()) + ":" + str(os.getgid()),
|
|
"--mount=type=bind,source=" + str(state) + ",target=/tmp/metis-audit-state.json,readonly"]
|
|
args.append("--storage-state=/tmp/metis-audit-state.json")
|
|
# Reutiliza el cliente de diagnóstico existente, sin añadir un SDK ni duplicar JSON-RPC.
|
|
client_type = runpy.run_path(str(ROOT / "scripts/check-playwright-mcp.py"))["StdioClient"]
|
|
client = client_type(args, 120)
|
|
try:
|
|
client.request("initialize", {"protocolVersion": "2025-06-18", "capabilities": {},
|
|
"clientInfo": {"name": "metis-authenticated-audit", "version": "1"}})
|
|
client.send({"method": "notifications/initialized"})
|
|
yield client
|
|
finally:
|
|
try:
|
|
client.close()
|
|
finally:
|
|
subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=15, check=False)
|
|
|
|
|
|
def capture(client, directory, name):
|
|
result = tool(client, "browser_take_screenshot", {"type": "png", "fullPage": False, "scale": "css"})
|
|
part = next(item for item in result["content"] if item.get("type") == "image" and item.get("mimeType") == "image/png")
|
|
data = base64.b64decode(part["data"], validate=True)
|
|
if not data.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
raise RuntimeError("Captura PNG inválida.")
|
|
destination = directory / (name + ".png")
|
|
destination.write_bytes(data)
|
|
return {"file": destination.name, "sha256": hashlib.sha256(data).hexdigest()}
|
|
|
|
|
|
def verify_browser(config, origin, fixture, directory, state, logout):
|
|
code_tool = "browser_run_code_unsafe"
|
|
evidence = []
|
|
assets_path = "/companies/" + fixture["company"] + "/assets"
|
|
with browser(config) as client:
|
|
print("Verificando el formulario de acceso real…", flush=True)
|
|
tool(client, "browser_resize", {"width": 1280, "height": 900})
|
|
tool(client, "browser_navigate", {"url": origin + "/login"})
|
|
tool(client, code_tool, {"code": "async (page) => { await page.getByRole('button', {name:'Entrar', exact:true}).waitFor(); }"})
|
|
evidence.append(capture(client, directory, "00-login"))
|
|
tool(client, "browser_fill_form", {"fields": [
|
|
{"name": "Correo electrónico", "type": "textbox", "target": "input[type=email]", "value": fixture["email"]},
|
|
{"name": "Contraseña", "type": "textbox", "target": "input[type=password]", "value": fixture["password"]},
|
|
]})
|
|
tool(client, "browser_click", {"target": "button[type=submit]"})
|
|
tool(client, code_tool, {"code": "async (page) => { await page.locator('[data-slot=workspace]').waitFor({timeout:30000}); }"})
|
|
identity = result_json(tool(client, "browser_evaluate", {"function": """async () => {
|
|
const response = await fetch('/api/auth/session');
|
|
const {user} = await response.json();
|
|
return {id:user?.id, admin:user?.isInitialAdmin, usersStatus:(await fetch('/api/users')).status};
|
|
}"""}))
|
|
if identity != {"id": fixture["user"], "admin": False, "usersStatus": 403}:
|
|
raise RuntimeError("La sesión no corresponde al usuario ordinario esperado.")
|
|
evidence.append(capture(client, directory, "01-authenticated"))
|
|
print("Usuario ordinario autenticado; verificando listado, ficha y diálogo…", flush=True)
|
|
tool(client, "browser_navigate", {"url": origin + assets_path})
|
|
tool(client, code_tool, {"code": "async (page) => { await page.getByRole('link', {name:'AUDIT-PORT-01', exact:true}).waitFor(); }"})
|
|
evidence.append(capture(client, directory, "02-assets"))
|
|
tool(client, code_tool, {"code": "async (page) => { await page.getByRole('link', {name:'AUDIT-PORT-01', exact:true}).click(); await page.getByRole('heading', {name:'AUDIT-PORT-01', exact:true}).waitFor(); }"})
|
|
evidence.append(capture(client, directory, "03-detail"))
|
|
tool(client, code_tool, {"code": "async (page) => { await page.getByRole('button', {name:'Editar activo', exact:true}).click(); await page.getByRole('dialog', {name:'Editar activo', exact:true}).waitFor(); }"})
|
|
evidence.append(capture(client, directory, "04-dialog"))
|
|
tool(client, code_tool, {"code": "async (page) => { await page.getByRole('dialog').getByRole('button', {name:'Cancelar', exact:true}).click(); await page.getByRole('dialog').waitFor({state:'hidden'}); }"})
|
|
storage = result_json(tool(client, code_tool, {"code": "async (page) => await page.context().storageState()"}))
|
|
private_write(state, json.dumps(storage))
|
|
# Otro contenedor comprueba la reutilización por el mismo mecanismo que recibirá Codex.
|
|
with browser(config, state) as client:
|
|
tool(client, "browser_navigate", {"url": origin + assets_path})
|
|
tool(client, code_tool, {"code": "async (page) => { await page.getByRole('link', {name:'AUDIT-PORT-01', exact:true}).waitFor(); }"})
|
|
if logout:
|
|
result = result_json(tool(client, "browser_evaluate", {"function": """async () => {
|
|
const {csrfToken} = await (await fetch('/api/auth/session')).json();
|
|
const response = await fetch('/api/auth/logout', {
|
|
method:'POST', body:'{}', headers:{'content-type':'application/json', 'x-csrf-token':csrfToken}
|
|
});
|
|
return {logout:response.status, protected:(await fetch('/api/companies')).status};
|
|
}"""}))
|
|
if result != {"logout": 200, "protected": 401}:
|
|
raise RuntimeError("El cierre de sesión no revocó el acceso esperado.")
|
|
return evidence
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--check", action="store_true", help="Verifica y retira el entorno; sin esta opción espera hasta Ctrl+C")
|
|
options = parser.parse_args()
|
|
for executable in ("docker", "node", "pnpm"):
|
|
if shutil.which(executable) is None:
|
|
parser.error("Falta " + executable + "; consulta docs/playwright-mcp.md")
|
|
if not command(["node", "--version"]).stdout.startswith("v24."):
|
|
parser.error("Se necesita Node.js 24.x.")
|
|
with (ROOT / ".codex/config.toml").open("rb") as source:
|
|
config = tomllib.load(source)["mcp_servers"]["playwright"]
|
|
images = [arg for arg in config["args"] if re.fullmatch(r"mcr\.microsoft\.com/playwright/mcp:v[\d.]+@sha256:[a-f0-9]{64}", arg)]
|
|
if config["command"] != "docker" or len(images) != 1 or "--network=host" not in config["args"]:
|
|
parser.error("Se requiere la configuración Docker fijada con red host.")
|
|
ARTIFACTS.mkdir(parents=True, exist_ok=True)
|
|
directory = Path(tempfile.mkdtemp(prefix="authenticated-", dir=ARTIFACTS))
|
|
container = "metis-visual-db-" + uuid.uuid4().hex[:12]
|
|
container_created = False
|
|
database_password = secrets.token_urlsafe(32)
|
|
api_port, web_port = available_port(), available_port()
|
|
while api_port == web_port:
|
|
web_port = available_port()
|
|
origin = "http://localhost:" + str(web_port)
|
|
api = TestApi("http://127.0.0.1:" + str(api_port), origin)
|
|
state = directory / "session.json"
|
|
env_file = directory / "database.env"
|
|
try:
|
|
private_write(env_file, "MARIADB_DATABASE=metis_desk_test\nMARIADB_USER=visual_audit\nMARIADB_PASSWORD=" + database_password + "\nMARIADB_ROOT_PASSWORD=" + secrets.token_urlsafe(32) + "\n")
|
|
print("Preparando MariaDB desechable y servicios en puertos libres…", flush=True)
|
|
command(["docker", "run", "--detach", "--rm", "--name=" + container,
|
|
"--label=metis.visual-audit=true", "--publish=127.0.0.1::3306",
|
|
"--env-file=" + str(env_file), DATABASE_IMAGE])
|
|
container_created = True
|
|
env_file.unlink()
|
|
wait_for(lambda: subprocess.run(["docker", "exec", container, "healthcheck.sh", "--connect", "--innodb_initialized"],
|
|
capture_output=True, timeout=15, check=False).returncode == 0)
|
|
bindings = json.loads(command(["docker", "inspect", "--format={{json .NetworkSettings.Ports}}", container]).stdout)
|
|
database_port = bindings["3306/tcp"][0]["HostPort"]
|
|
# No carga .env: las únicas credenciales apuntan al contenedor recién creado.
|
|
environment = {**os.environ, "DB_HOST": "127.0.0.1", "DB_PORT": database_port,
|
|
"DB_NAME": "metis_desk_test", "DB_USER": "visual_audit", "DB_PASSWORD": database_password,
|
|
"SESSION_SECRET": secrets.token_urlsafe(48), "NODE_ENV": "test", "APP_ORIGIN": origin,
|
|
"PORT": str(api_port), "API_PROXY_TARGET": api.base}
|
|
with child_process(["node", "--import", "tsx", "apps/server/src/index.ts"], environment, directory / "api.log") as backend:
|
|
wait_for(api.ready, backend)
|
|
fixture = prepare_fixture(api)
|
|
TestApi(api.base, origin).request("/companies", expected=401)
|
|
with child_process(["pnpm", "--config.verify-deps-before-run=false", "--filter", "@metis/web", "exec", "vite",
|
|
"--host", "127.0.0.1", "--port", str(web_port), "--strictPort"], environment, directory / "vite.log") as frontend:
|
|
web = TestApi(origin, origin)
|
|
wait_for(web.ready, frontend)
|
|
evidence = verify_browser(config, origin, fixture, directory, state, options.check)
|
|
report = {"origin": origin, "companyId": fixture["company"], "assetId": fixture["asset"],
|
|
"ordinaryUser": True, "resources": fixture["resources"],
|
|
"unauthenticatedStatus": 401, "userManagementStatus": 403,
|
|
"reusedSession": True, "logoutChecked": options.check,
|
|
"databaseImage": DATABASE_IMAGE, "mcpImage": images[0], "screenshots": evidence}
|
|
(directory / "report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
|
|
print("Verificado: login real, usuario ordinario, listado, ficha, diálogo cancelado y sesión en otro contenedor.", flush=True)
|
|
print("Evidencia local: " + str(directory.relative_to(ROOT)), flush=True)
|
|
if not options.check:
|
|
print("Mantén este proceso abierto. En otra terminal WSL:", flush=True)
|
|
print("METIS_PLAYWRIGHT_STORAGE_STATE=" + str(state.relative_to(ROOT)) + " bash scripts/codex-playwright-wsl.sh", flush=True)
|
|
print("URL de auditoría: " + origin + "/companies/" + fixture["company"] + "/assets", flush=True)
|
|
print("Ctrl+C retira exclusivamente este entorno y su sesión.", flush=True)
|
|
signal.pause()
|
|
finally:
|
|
env_file.unlink(missing_ok=True)
|
|
state.unlink(missing_ok=True)
|
|
if container_created:
|
|
removed = subprocess.run(["docker", "rm", "-f", "-v", container], capture_output=True, timeout=30, check=False)
|
|
if removed.returncode != 0 and "No such container" not in removed.stderr.decode(errors="replace"):
|
|
raise RuntimeError("No se pudo confirmar la retirada del contenedor temporal; consulta la limpieza manual documentada.")
|
|
print("Entorno temporal retirado; las capturas se conservan sin sesión ni credenciales.", flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
signal.signal(signal.SIGTERM, lambda *_: sys.exit(130))
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
print("Auditoría finalizada.")
|
|
except (OSError, RuntimeError, ValueError, KeyError, StopIteration, subprocess.SubprocessError) as error:
|
|
# No volcar excepciones de HTTP/MCP: podrían incluir contraseñas o cookies.
|
|
print("La comprobación falló. Revisa los requisitos Docker/WSL y los logs de servicios locales; no se ha validado la integración.", file=sys.stderr)
|
|
if isinstance(error, RuntimeError):
|
|
print(str(error), file=sys.stderr)
|
|
sys.exit(1)
|