دليل مزود REST

نظرة سريعة

ما: استخدم مزود rest المدمج لفحوصات الأدلة المحدودة GET. لماذا: قم ببوابة مطالبات API البعيدة دون بناء مزود MCP خارجي. من: المشغلون والمندمجون الذين يؤلفون شروط مدعومة بـ REST. المتطلبات المسبقة: getting_started.md, condition_authoring.md

---

النطاق والضمانات

مزود REST V1 ضيق عن عمد:

• الحصول فقط
• checks: json_path, header
• فشل التحقق من صحة الطلب/الاستجابة المغلقة
• انبعاث مرساة دليل حتمي (rest_request)

سلوك حرج للأمان:

• json_path يتطلب نوع محتوى JSON (application/json أو *+json)
• لا يمكن لرؤوس الاستعلام تجاوز الرؤوس المحجوزة / المدارة بواسطة المصادقة
• يتم رفض إعادة التوجيه
• يتم فرض حدود حجم الاستجابة ووقت الانتظار

امتداد الفجوة (لمتكاملين المنصة):

• OSS runtime exposes RestPolicyEvaluator hooks for:
  • قرارات مضيف/نظام الخروج
  • قرارات نظام المصادقة
• يقوم مقيم OSS الافتراضي بالحفاظ على السلوك السابق (السماح).
• يمكن للمقيمين المخصصين الرفض باستخدام رموز أسباب حتمية للتدقيق/الحوكمة.

---

تكوين المزود

تسجيل مدمج بسيط:

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

مع المصادقة والرؤوس الافتراضية:

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

مثال على التحقق من صحة التكوين:

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

---

أمثلة على الشروط

json_path الشرط:

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

---

استعلام دليل قابل للتنفيذ

هذا الكتلة تقوم بتشغيل نموذج HTTP محلي، تستدعي evidence_query ضد مزود rest، وتؤكد على قيم الحقول الثابتة + الحقول المرتبطة.

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)

---

ملاحظات

• استخدم evidence_query لتصحيح الأخطاء على مستوى المزود وفحص الأخطاء بشكل حتمي.
• لنتائج البوابة، قم بتشغيل scenario_next وتفقد آثار التشغيل عند الحاجة.
• يُفضل استخدام قوائم السماح الصريحة للمضيفين والحفاظ على allow_private_networks = false ما لم يكن ذلك مطلوبًا.