Grabadora Asset Core Puerta de Decisión
Docs
Documentos de Decision Gate

Evaluación de puertas determinista, reproducible con decisiones auditables.

Documentación de Asset Core

Manual del Proveedor REST

A primera vista

Qué: Utilice el proveedor rest integrado para verificaciones de evidencia GET limitadas. Por qué: Controle las reclamaciones de API remotas sin construir un proveedor MCP externo. Quién: Operadores e integradores que crean condiciones respaldadas por REST. Requisitos previos: getting_started.md, condition_authoring.md

Alcance y Garantías

El proveedor REST V1 es intencionalmente limitado:

  • SOLO GET
  • checks: json_path, header
  • validación de solicitud/respuesta que falla en cerrado
  • emisión de anclaje de evidencia determinista (rest_request)

Comportamiento crítico para la seguridad:

  • json_path requiere tipo de contenido JSON (application/json o *+json)
  • los encabezados de consulta no pueden anular los encabezados reservados/gestionados por autenticación
  • las redirecciones son rechazadas
  • se aplican límites de tamaño de respuesta y tiempo de espera

Costura de extensibilidad (para integradores de plataforma):

  • OSS runtime exposes RestPolicyEvaluator hooks for:
    • decisiones de host/esquema de salida
    • decisiones del esquema de autenticación
  • El evaluador OSS predeterminado preserva el comportamiento anterior (permitir).
  • Los evaluadores personalizados pueden denegar con códigos de razón deterministas para auditoría/gobernanza.

Configuración del Proveedor

Registro mínimo incorporado:

[[providers]]
name = "rest"
type = "builtin"
config = { allow_http = false, timeout_ms = 5000, max_response_bytes = 1048576, allowed_hosts = ["api.example.com"], allow_private_networks = false, user_agent = "decision-gate/0.1", hash_algorithm = "sha256" }

Con autenticación y encabezados predeterminados:

[[providers]]
name = "rest"
type = "builtin"
allow_raw = true
config = { allow_http = false, timeout_ms = 5000, max_response_bytes = 1048576, allowed_hosts = ["api.example.com"], allow_private_networks = false, user_agent = "decision-gate/0.1", hash_algorithm = "sha256", auth = { bearer_token = "${env:API_TOKEN}" }, default_headers = { x_dg_client = "decision-gate" } }

Ejemplo de validación de configuración:

[server]
transport = "http"
bind = "127.0.0.1:4000"
mode = "strict"

[server.auth]
mode = "local_only"

[namespace]
allow_default = true
default_tenants = [1]

[trust]
default_policy = "audit"
min_lane = "verified"

[evidence]
allow_raw_values = true
require_provider_opt_in = true

[schema_registry]
type = "sqlite"
path = "decision-gate-registry.db"

[schema_registry.acl]
allow_local_only = true
require_signing = false

[run_state_store]
type = "sqlite"
path = "decision-gate.db"
journal_mode = "wal"
sync_mode = "full"
busy_timeout_ms = 5000

[[providers]]
name = "time"
type = "builtin"

[[providers]]
name = "env"
type = "builtin"

[[providers]]
name = "json"
type = "builtin"
config = { root = "./evidence", root_id = "evidence-root", max_bytes = 1048576, allow_yaml = true }

[[providers]]
name = "http"
type = "builtin"

[[providers]]
name = "rest"
type = "builtin"
allow_raw = true
config = { allow_http = true, timeout_ms = 5000, max_response_bytes = 1048576, allowed_hosts = ["127.0.0.1"], allow_private_networks = true, user_agent = "decision-gate/0.1", hash_algorithm = "sha256" }

Ejemplos de Condiciones

json_path condición:

{
  "condition_id": "remote_approved",
  "query": {
    "provider_id": "rest",
    "check_id": "json_path",
    "params": {
      "url": "https://api.example.com/decision/42",
      "jsonpath": "$.approved",
      "headers": { "x-client": "dg" }
    }
  },
  "comparator": "equals",
  "expected": true,
  "policy_tags": []
}

header condición:

{
  "condition_id": "remote_etag_present",
  "query": {
    "provider_id": "rest",
    "check_id": "header",
    "params": {
      "url": "https://api.example.com/decision/42",
      "header_name": "etag"
    }
  },
  "comparator": "exists",
  "expected": null,
  "policy_tags": []
}

Consulta de Evidencia Ejecutable

Este bloque inicia un fixture HTTP local, llama a evidence_query contra el proveedor rest, y afirma los campos de determinismo de valor + ancla.

import json
import os
import threading
from http.server import BaseHTTPRequestHandler
from http.server import HTTPServer
from urllib import request


class Handler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        if self.path != "/decision":
            self.send_response(404)
            self.end_headers()
            return
        body = json.dumps({"approved": True, "summary": {"count": 7}}).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, _format: str, *_args: object) -> None:
        return


def call_tool(endpoint: str, tool_name: str, arguments: dict) -> dict:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {"name": tool_name, "arguments": arguments},
    }
    req = request.Request(
        endpoint,
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with request.urlopen(req, timeout=10) as resp:
        body = json.loads(resp.read().decode("utf-8"))
    if "error" in body:
        raise RuntimeError(f"tool call failed: {body['error']}")
    content = body.get("result", {}).get("content", [])
    if not content or "json" not in content[0]:
        raise RuntimeError(f"unexpected tool result envelope: {body}")
    return content[0]["json"]


server = HTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()

try:
    endpoint = os.environ.get("DG_ENDPOINT", "http://127.0.0.1:8080/rpc")
    fixture_url = f"http://127.0.0.1:{server.server_port}/decision"
    response = call_tool(
        endpoint,
        "evidence_query",
        {
            "query": {
                "provider_id": "rest",
                "check_id": "json_path",
                "params": {"url": fixture_url, "jsonpath": "$.approved"},
            },
            "context": {
                "tenant_id": 1,
                "namespace_id": 1,
                "run_id": "docs-run-1",
                "scenario_id": "docs-scenario",
                "stage_id": "main",
                "trigger_id": "docs-trigger-1",
                "trigger_time": {"kind": "logical", "value": 1},
                "correlation_id": None,
            },
        },
    )

    result = response["result"]
    assert result["error"] is None, result
    assert result["value"] == {"kind": "json", "value": True}, result
    anchor = result["evidence_anchor"]
    assert anchor["anchor_type"] == "rest_request", anchor
    anchor_value = json.loads(anchor["anchor_value"])
    assert anchor_value["check_id"] == "json_path", anchor_value
    assert "response_body_hash" in anchor_value, anchor_value
finally:
    server.shutdown()
    thread.join(timeout=5)

Notas

  • Utiliza evidence_query para la depuración a nivel de proveedor y la inspección de errores deterministas.
  • Para los resultados de la puerta, ejecuta scenario_next e inspecciona los artefactos de runpack cuando sea necesario.
  • Prefiera listas de permitidos de host explícitas y mantenga allow_private_networks = false a menos que sea necesario.