Gravadora Asset Core Decision Gate
Docs
Decision Gate Docs

Avaluació de portes determinista, reproduïble amb decisions auditable.

Asset Core docs

Manual del Proveïdor REST

A Simple Vista

Què: Utilitzeu el proveïdor rest integrat per a comprovacions d’evidència GET limitades. Per què: Controlar les reclamacions de l’API remota sense construir un proveïdor MCP extern. Qui: Operadors i integradors que redacten condicions basades en REST. Requisits previs: getting_started.md, condition_authoring.md

Àmbit i Garanties

El proveïdor REST V1 és intencionadament estret:

  • NOMÉS GET
  • comprovacions: json_path, header
  • validació de sol·licituds/respostes que falla tancada
  • emissió d’ancoratge d’evidència determinista (rest_request)

Comportament crític per a la seguretat:

  • json_path requereix el tipus de contingut JSON (application/json o *+json)
  • les capçaleres de consulta no poden substituir les capçaleres reservades/gestionades per l’autenticació
  • les redireccions són rebutjades
  • s’apliquen les limitacions de mida de resposta i temps d’espera

Seam d’extensibilitat (per a integradors de plataforma):

  • OSS runtime exposes RestPolicyEvaluator hooks for:
    • decisions d’amfitrió/esquema d’egress
    • decisions d’esquema d’autenticació
  • L’evaluador OSS per defecte preserva el comportament anterior (permet).
  • Els avaluadors personalitzats poden denegar amb codis de raó deterministes per a auditoria/governança.

Configuració del proveïdor

Registre mínim integrat:

[[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" }

Amb autenticació i capçaleres per defecte:

[[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" } }

Exemple de validació de configuració:

[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" }

Exemples de Condicions

json_path condició:

{
  "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ó:

{
  "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 d’Evidència Executable

Aquest bloc inicia un fixture HTTP local, crida evidence_query contra el proveïdor rest, i afirma els camps de valor + ancoratge de determinisme.

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)

Notes

  • Utilitzeu evidence_query per a la depuració a nivell de proveïdor i la inspecció d’errors deterministes.
  • Per als resultats de la porta, executa scenario_next i inspecciona els artefactes de runpack quan sigui necessari.
  • Preferiu llistes d’hostes explícites i manteniu allow_private_networks = false a menys que sigui necessari.