Recorder Asset Core Decision Gate
Docs
Decision Gate Docs

Deterministic, replayable gate evaluation with auditable decisions.

Asset Core docs

REST Provider Playbook

At a Glance

What: Use the built-in rest provider for bounded GET evidence checks. Why: Gate remote API claims without building an external MCP provider. Who: Operators and integrators authoring REST-backed conditions. Prerequisites: getting_started.md, condition_authoring.md

Scope and Guarantees

REST provider V1 is intentionally narrow:

  • GET only
  • checks: json_path, header
  • fail-closed request/response validation
  • deterministic evidence anchor emission (rest_request)

Security-critical behavior:

  • json_path requires JSON content type (application/json or *+json)
  • query headers cannot override reserved/auth-managed headers
  • redirects are rejected
  • response size and timeout bounds are enforced

Extensibility seam (for platform integrators):

  • OSS runtime exposes RestPolicyEvaluator hooks for:
    • egress host/scheme decisions
    • auth-scheme decisions
  • Default OSS evaluator preserves prior behavior (allow).
  • Custom evaluators can deny with deterministic reason codes for audit/governance.

Provider Configuration

Minimal built-in registration:

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

With auth and default headers:

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

Config validation example:

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

Condition Examples

json_path condition:

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

{
  "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": []
}

Runnable Evidence Query

This block spins up a local HTTP fixture, calls evidence_query against the rest provider, and asserts value + anchor determinism fields.

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

  • Use evidence_query for provider-level debugging and deterministic error inspection.
  • For gate outcomes, run scenario_next and inspect runpack artifacts when needed.
  • Prefer explicit host allowlists and keep allow_private_networks = false unless required.