Building An Offline Llm Risk Gate With Signed Policy Artifacts

The problem I ran into: “The model was fine, the policy wasn’t”

I once shipped an AI feature where the prompt looked safe in testing, but in production a different “policy flavor” slipped in via a pipeline mismatch. The model didn’t suddenly become evil—what changed was the enforcement boundary: the system that was supposed to constrain outputs wasn’t reliably bound to the exact model + ruleset version.

That’s when I decided to build a tiny but strict “AI Risk Gate” that runs offline (no network calls) and only allows responses if they pass:

• a risk-policy decision (simple heuristics first, then extensible),
• a cryptographic policy artifact check (signed policy so it can’t be swapped),
• an allowlisted model identity (so prompts can’t be paired with the wrong model/rules),
• and a tamper-evident audit log (so I could prove what happened later).

Below is what I built and how it works end-to-end.

---

What I mean by “AI Trust” in practice

For me, “AI trust” wasn’t about magically making the model correct. It was about controlling conditions:

• Which policy was used?
• Which model was allowed?
• What decision was made?
• Can we detect tampering after the fact?

To do that, I focused on risk & security management around LLM outputs, using:

• signed policy artifacts (integrity/authenticity),
• offline evaluation (reproducibility),
• structured decisions (auditability).

---

Architecture at a glance

I implemented three pieces:

1. Policy artifact (policy.json)
   Contains allowed models + risk rules. This is signed.

2. Signature verification + decision engine (risk_gate.py)
   Offline script that:

   • verifies the signature,
   • confirms the model is allowed,
   • scores the candidate output for risk patterns,
   • emits an allow/deny decision.

3. Tamper-evident audit log
   Each decision event includes a hash chain so later edits are detectable.

---

Step 1: Create a signed policy artifact

Policy content: policy.json

This policy is intentionally small and specific: it denies outputs containing certain “exfiltration-like” patterns and blocks “command execution” language.

{
  "policy_version": "2026-08-15",
  "allowed_models": [
    { "id": "local-llm-7b", "min_revision": "r120" }
  ],
  "risk_rules": [
    {
      "id": "deny_command_execution_phrases",
      "type": "deny_pattern",
      "patterns": [
        "run this command",
        "execute the following",
        "sudo ",
        "chmod ",
        "curl http",
        "wget http"
      ]
    },
    {
      "id": "deny_secret_like_patterns",
      "type": "deny_pattern",
      "patterns": [
        "BEGIN PRIVATE KEY",
        "-----BEGIN",
        "api_key=",
        "token=",
        "password="
      ]
    },
    {
      "id": "high_risk_word_bonus",
      "type": "scoring_pattern",
      "patterns": [
        { "text": "exfiltrate", "score": 15 },
        { "text": "leak", "score": 10 },
        { "text": "bypass", "score": 12 }
      ]
    }
  ],
  "decision_thresholds": {
    "deny_if_any_deny_pattern_matches": true,
    "deny_if_risk_score_gte": 20
  }
}

Sign the policy: sign_policy.py

I used RSA with SHA-256. The keypair is created once, then the policy artifact is signed.

import json
from pathlib import Path
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa

POLICY_PATH = Path("policy.json")
PRIVATE_KEY_PATH = Path("risk_gate_private.pem")
PUBLIC_KEY_PATH = Path("risk_gate_public.pem")

def generate_keys():
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    public_key = private_key.public_key()

    PRIVATE_KEY_PATH.write_bytes(
        private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        )
    )

    PUBLIC_KEY_PATH.write_bytes(
        public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        )
    )

def sign_policy():
    policy_bytes = POLICY_PATH.read_bytes()
    private_key = serialization.load_pem_private_key(PRIVATE_KEY_PATH.read_bytes(), password=None)

    signature = private_key.sign(
        policy_bytes,
        padding.PKCS1v15(),
        hashes.SHA256()
    )

    Path("policy.sig").write_bytes(signature)
    print("Signed policy.json -> policy.sig")

if __name__ == "__main__":
    # Generate keys once; comment out after first run.
    # generate_keys()
    sign_policy()

Why I chose signing here: without signing, an attacker (or a CI bug) could swap policy.json while keeping the app unchanged. With signing, the gate refuses any policy that doesn’t match the embedded public key.

---

Step 2: The offline risk gate

Install dependency

This uses the cryptography library.

pip install cryptography

risk_gate.py (decision engine + audit chain)

import base64
import hashlib
import json
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

POLICY_PATH = Path("policy.json")
SIG_PATH = Path("policy.sig")
PUBLIC_KEY_PATH = Path("risk_gate_public.pem")

AUDIT_LOG_PATH = Path("audit.log.jsonl")


@dataclass
class Decision:
    allowed: bool
    risk_score: int
    deny_reasons: List[str]
    matched_patterns: List[str]
    policy_version: str
    model_id: str
    model_revision: str
    audit_event_hash: str
    previous_event_hash: str


def sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def verify_policy_signature(policy_bytes: bytes, signature: bytes) -> None:
    public_key = serialization.load_pem_public_key(PUBLIC_KEY_PATH.read_bytes())

    public_key.verify(
        signature,
        policy_bytes,
        padding.PKCS1v15(),
        hashes.SHA256()
    )
    # If verification fails, an exception is raised.


def load_policy() -> Dict[str, Any]:
    policy_bytes = POLICY_PATH.read_bytes()
    signature = SIG_PATH.read_bytes()
    verify_policy_signature(policy_bytes, signature)
    return json.loads(policy_bytes.decode("utf-8"))


def model_allowed(policy: Dict[str, Any], model_id: str, model_revision: str) -> bool:
    for m in policy["allowed_models"]:
        if m["id"] != model_id:
            continue
        # Simple "revision floor" check; in a real setup I’d use semantic versioning.
        return model_revision >= m["min_revision"]
    return False


def evaluate_risk(policy: Dict[str, Any], candidate_output: str) -> Dict[str, Any]:
    deny_reasons = []
    matched_patterns = []
    risk_score = 0

    for rule in policy["risk_rules"]:
        if rule["type"] == "deny_pattern":
            for p in rule["patterns"]:
                if p.lower() in candidate_output.lower():
                    deny_reasons.append(rule["id"])
                    matched_patterns.append(p)
            continue

        if rule["type"] == "scoring_pattern":
            for item in rule["patterns"]:
                text = item["text"]
                score = item["score"]
                if text.lower() in candidate_output.lower():
                    risk_score += score
                    matched_patterns.append(text)
            continue

    thresholds = policy["decision_thresholds"]
    allowed = True

    if thresholds.get("deny_if_any_deny_pattern_matches", True) and deny_reasons:
        allowed = False

    if allowed and risk_score >= thresholds.get("deny_if_risk_score_gte", 999999):
        allowed = False

    return {
        "allowed": allowed,
        "risk_score": risk_score,
        "deny_reasons": deny_reasons,
        "matched_patterns": matched_patterns
    }


def load_last_audit_hash() -> str:
    if not AUDIT_LOG_PATH.exists() or AUDIT_LOG_PATH.stat().st_size == 0:
        return "GENESIS"
    last_line = AUDIT_LOG_PATH.read_text(encoding="utf-8").splitlines()[-1]
    last_event = json.loads(last_line)
    return last_event["audit_event_hash"]


def append_audit_event(event: Dict[str, Any]) -> str:
    prev_hash = load_last_audit_hash()

    # Hash chain: include previous hash so edits are detectable.
    payload = json.dumps(event, sort_keys=True).encode("utf-8")
    audit_event_hash = sha256_hex(prev_hash.encode("utf-8") + payload)

    record = dict(event)
    record["previous_event_hash"] = prev_hash
    record["audit_event_hash"] = audit_event_hash

    with AUDIT_LOG_PATH.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

    return audit_event_hash


def gate(model_id: str, model_revision: str, candidate_output: str) -> Decision:
    policy = load_policy()

    if not model_allowed(policy, model_id, model_revision):
        event = {
            "timestamp": time.time(),
            "model_id": model_id,
            "model_revision": model_revision,
            "policy_version": policy["policy_version"],
            "candidate_output_preview": candidate_output[:120],
            "decision": "deny",
            "reason": "model_not_allowed"
        }
        audit_hash = append_audit_event(event)
        return Decision(
            allowed=False,
            risk_score=0,
            deny_reasons=["model_not_allowed"],
            matched_patterns=[],
            policy_version=policy["policy_version"],
            model_id=model_id,
            model_revision=model_revision,
            audit_event_hash=audit_hash,
            previous_event_hash=load_last_audit_hash()
        )

    evaluation = evaluate_risk(policy, candidate_output)

    decision_str = "allow" if evaluation["allowed"] else "deny"
    event = {
        "timestamp": time.time(),
        "model_id": model_id,
        "model_revision": model_revision,
        "policy_version": policy["policy_version"],
        "candidate_output_preview": candidate_output[:120],
        "decision": decision_str,
        "risk_score": evaluation["risk_score"],
        "deny_reasons": evaluation["deny_reasons"],
        "matched_patterns": evaluation["matched_patterns"]
    }

    audit_hash = append_audit_event(event)

    return Decision(
        allowed=evaluation["allowed"],
        risk_score=evaluation["risk_score"],
        deny_reasons=evaluation["deny_reasons"],
        matched_patterns=evaluation["matched_patterns"],
        policy_version=policy["policy_version"],
        model_id=model_id,
        model_revision=model_revision,
        audit_event_hash=audit_hash,
        previous_event_hash=load_last_audit_hash()
    )


if __name__ == "__main__":
    # Example run (offline).
    # Candidate output simulates what an LLM might generate.
    candidate = "Sure, run this command: sudo rm -rf /"

    d = gate(
        model_id="local-llm-7b",
        model_revision="r130",
        candidate_output=candidate
    )

    print("=== Decision ===")
    print(json.dumps({
        "allowed": d.allowed,
        "risk_score": d.risk_score,
        "deny_reasons": d.deny_reasons,
        "matched_patterns": d.matched_patterns,
        "policy_version": d.policy_version,
        "audit_event_hash": d.audit_event_hash
    }, indent=2))

What each important block is doing (and why)

• verify_policy_signature: checks that policy.json matches policy.sig using the pinned public key. If someone changes the policy, the gate refuses to run rules.
• model_allowed: ensures the incoming model_id + model_revision pair matches the allowlist floors in the signed policy.
• evaluate_risk: applies two categories of rules:
  • deny_pattern rules: any match immediately denies.
  • scoring_pattern rules: matches add to a risk score; exceeding a threshold denies.
• append_audit_event: writes a JSON line event to audit.log.jsonl where each event hash includes the previous event hash. This forms a tamper-evident chain.

---

Step 3: Run it with a “safe” and an “unsafe” output

Safe example

Replace candidate with:

candidate = "I can explain how to rotate API keys safely without revealing secrets."

Expected behavior:

• No deny patterns match.
• No “exfiltrate/leak/bypass” scoring patterns match.
• Decision should be allow.

Unsafe example (the included one)

"Sure, run this command: sudo rm -rf /"

Expected behavior:

• Deny pattern triggers ("sudo " and "run this command").
• Decision should be deny.

Running the script prints the decision and appends an event to audit.log.jsonl.

Example audit log lines look like:

{
  "timestamp": 1720000000.0,
  "model_id": "local-llm-7b",
  "model_revision": "r130",
  "policy_version": "2026-08-15",
  "candidate_output_preview": "Sure, run this command: sudo rm -rf /",
  "decision": "deny",
  "risk_score": 0,
  "deny_reasons": ["deny_command_execution_phrases"],
  "matched_patterns": ["run this command", "sudo "],
  "previous_event_hash": "GENESIS",
  "audit_event_hash": "..."
}

---

A quick tamper check I used

When I needed to validate audit integrity, I re-hashed the chain by reading audit.log.jsonl top to bottom and checking that each stored audit_event_hash matches the computed one using the previous hash.

That’s a practical “trust but verify” loop: even if someone gains access later, the log history becomes fragile to edits.

---

Security takeaways I actually cared about

1. Signed policies are the boundary, not the prompt. I learned that prompt safety isn’t enough if policy enforcement can drift.
2. Offline evaluation reduces non-determinism. When the gate runs without fetching external dependencies, I can reproduce decisions reliably.
3. Audit trails need tamper evidence, not just timestamps. Hash chaining made the log “detectably wrong” instead of “quietly wrong”.
4. Model identity must be part of the enforcement context. Allowlisting the model ID + revision prevented mismatched rule application.

---

Conclusion

I built an offline AI Risk Gate that only allows LLM outputs when a signed policy artifact verifies correctly, the model identity is allowlisted, and the output passes deterministic risk rules. The gate also produces a tamper-evident audit log using a hash chain, turning “we think it was safe” into “we can prove what policy decided what outcome.”