hexarch-guardrails 0.4.5

Lightweight policy-driven API protection and guardrails library

Hexarch Guardrails Python SDK

Stop production disasters before they happen. Lightweight policy-driven API protection for students, solo developers, and hackathons.

🚨 Stop Disasters Before They Happen

Has this ever happened to you?

# Innocent cleanup script...
def cleanup_old_data():
    db.execute("DELETE FROM users WHERE last_login < '2023-01-01'")

cleanup_old_data()  # 💥 Just deleted 10,000 production records

Or this?

# Weekend GPT-4 experiment...
for prompt in user_prompts:  # 1000 prompts
    openai.ChatCompletion.create(model="gpt-4", ...)
    
# Monday morning: $4,200 OpenAI bill 💸

With Hexarch Guardrails:

from hexarch_guardrails import Guardian
guardian = Guardian()

@guardian.check("safe_delete")  # ← Add one line
def cleanup_old_data():
    db.execute("DELETE FROM users WHERE last_login < '2023-01-01'")

# Now it blocks: ❌ [BLOCKED] Policy 'safe_delete' requires confirmation

→ Try the 30-second interactive demo

---

Source-of-truth

This SDK is synced from the private monorepo (no1rstack/Hexarch) via an automated subtree publish workflow.

Installation

pip install hexarch-guardrails

Optional Extras (Install by Feature)

Feature	Install Command
Admin CLI (hexarch-ctl)	pip install hexarch-guardrails[cli]
REST API Server	pip install "hexarch-guardrails[server]"
Postgres-backed storage	pip install "hexarch-guardrails[postgres]"
Dev/Test tooling	pip install "hexarch-guardrails[dev]"

Quick Start

1. Create a policy file (hexarch.yaml)

policies:
  - id: "api_budget"
    description: "Protect against overspending"
    rules:
      - resource: "openai"
        monthly_budget: 10
        action: "warn_at_80%"

  - id: "rate_limit"
    description: "Prevent API abuse"
    rules:
      - resource: "*"
        requests_per_minute: 100
        action: "block"

  - id: "safe_delete"
    description: "Require confirmation for destructive ops"
    rules:
      - operation: "delete"
        action: "require_confirmation"

2. Use in your code

from hexarch_guardrails import Guardian

# Initialize (auto-discovers hexarch.yaml)
# Optional: pass an explicit path if auto-discovery fails
guardian = Guardian(policy_file="/absolute/path/to/hexarch.yaml")

# Protect API calls
@guardian.check("api_budget")
def call_openai(prompt):
    import openai
    return openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

# Use it
response = call_openai("Hello AI!")

---

How Enforcement Actually Works (No Surprises)

1) Confirmation is not a CLI prompt

Guardrails does not pause for STDIN or interactive prompts. If a policy requires confirmation, you must handle confirmation in your app and only call the guarded function after confirmation is obtained.

Example pattern (explicit context flag):

from hexarch_guardrails import Guardian
from hexarch_guardrails.exceptions import PolicyViolation

guardian = Guardian()

# Require explicit confirmation in policy via input.confirm == true
@guardian.check("safe_delete", context={"confirm": True})
def delete_user_data(user_id: str):
    db.delete(user_id)

try:
    delete_user_data("user_123")
except PolicyViolation as exc:
    print(f"Blocked: {exc}")

If you need a human-in-the-loop confirmation for a headless script, collect it before calling the guarded function (CLI flag, UI confirmation, or workflow approval).

2) What happens when a policy blocks

Guardrails blocks by raising PolicyViolation before your function runs. Your app doesn't crash if you handle the exception.

from hexarch_guardrails.exceptions import PolicyViolation

try:
    call_openai("Hello AI!")
except PolicyViolation as exc:
    # Decide what to do (log, return fallback, notify, etc.)
    print(f"Denied: {exc}")

3) Budget + rate state storage

The core SDK delegates state to the policy engine (OPA by default). The Safe Automation Runner demo uses an in-process evaluator with in-memory state (non-distributed, demo only).

For production or distributed environments:

• Run OPA as a service and back it with persistent storage (e.g., OPA bundles + data API + external state store).
• Or replace the policy engine with your own decision service that tracks budgets centrally.

4) OPA dependency (what’s included vs external)

pip install hexarch-guardrails installs the SDK only. OPA is a separate service if you use the default evaluator. The demo uses a lightweight in-process evaluator to avoid external dependencies.

5) Auto-discovery vs explicit policy file path

Auto-discovery walks up from your current working directory to find hexarch.yaml. If your app runs in Docker or from a different working directory, pass the path explicitly:

guardian = Guardian(policy_file="/app/config/hexarch.yaml")

---

Confirmation Workflows: Three Proven Patterns

When a policy requires confirmation (e.g., requires_confirmation: true), your app must collect approval before retrying. Here are three production patterns:

Pattern 1: Interactive Confirmation (Development)

For: CLI tools, local development, interactive dashboards

Prompt user on console and block until they respond:

from hexarch_guardrails import ConfirmationWorkflow, MemoryConfirmationStore

workflow = ConfirmationWorkflow(guardian, MemoryConfirmationStore())

result = workflow.execute_with_confirmation(
    policy_id="sensitive_delete",
    func=delete_records,
    func_args=(record_ids,),
    interactive=True  # Prompts user: "Approve? (yes/no)"
)

Pros: Simple, no setup
Cons: Blocks caller, requires terminal access

Pattern 2: Token-Based Confirmation (Microservices)

For: Microservices, cron jobs, background tasks, distributed systems

Return a token immediately instead of blocking:

from hexarch_guardrails import MemoryConfirmationStore, PolicyViolation
import uuid

store = MemoryConfirmationStore()  # Or RedisConfirmationStore for production

try:
    guarded = guardian.check("sensitive_delete")(delete_records)
    result = guarded(record_ids)
except PolicyViolation as e:
    # Generate token and store approval request
    token = str(uuid.uuid4())
    store.store_pending(token, {"operation": "delete", "count": len(record_ids)})
    
    # Return token to caller (don't block)
    return {"status": "pending_approval", "token": token, "reason": str(e)}

# Later, when approved by external service:
if store.is_approved(token):
    context = {"confirmation_token": token}
    guarded = guardian.check("sensitive_delete", context)(delete_records)
    result = guarded(record_ids)

Pros: Non-blocking, scales to many concurrent requests
Cons: More complex, needs token persistence (Redis/DB)

Pattern 3: HTTP Callback Confirmation (Public APIs)

For: REST APIs, regulated environments, audit requirements

Delegate approval to external service and return 202 Accepted:

from hexarch_guardrails import HTTPConfirmationGateway, PolicyViolation

gateway = HTTPConfirmationGateway(
    webhook_url="https://approval-service.example.com/approve",
    headers={"Authorization": "Bearer token"}
)

@app.delete("/users/{user_id}")
def delete_user(user_id):
    try:
        guarded = guardian.check("user_deletion")(db.delete_user)
        result = guarded(user_id)
        return {"status": "deleted"}
    except PolicyViolation as e:
        # Notify external approval service
        token = gateway.notify_pending_approval(
            reason=str(e),
            operation="delete_user",
            metadata={"user_id": user_id}
        )
        
        # Return 202 Accepted (async operation)
        return {
            "status": 202,
            "token": token,
            "approval_url": f"https://approval.example.com/decisions/{token}",
            "retry_header": f"X-Confirmation-Token: {token}"
        }

# Client retries with token:
# DELETE /users/123
# X-Confirmation-Token: abc123def456

Pros: Scales to enterprise approval workflows (Slack, email, JIRA), audit trail
Cons: Most setup, external dependency

Choosing Your Pattern

Pattern	Local Dev	Cron/Background	REST API	Microservice
Interactive	✅	❌	❌	❌
Token	✅	✅	✅	✅
HTTP Callback	❌	❌	✅✅	✅

Token Persistence Options

• In-Memory (dev only): MemoryConfirmationStore()
• Redis (production): RedisConfirmationStore(redis.Redis())
• Database (custom): Implement ConfirmationStore interface
• Signed JWT (stateless): Pass token to client, validate signature on retry

Complete Working Example

See examples/confirmation_patterns/ for runnable demos of all three patterns. Run:

cd examples/confirmation_patterns
pip install -r requirements.txt
python main.py

Then explore:

• interactive_pattern.py - CLI confirmation with user prompt
• token_pattern.py - Async workflow with token persistence
• http_pattern.py - REST API with webhook callbacks

---

Production Patterns (Concrete How-To)

A) Confirmation in multi-step workflows (headless/cron-safe)

Guardrails does not store confirmation state for you. In production, treat confirmation as your application state (DB row, cache entry, signed token), then include it in the policy context when re-calling the function.

Example pattern (store confirmation token and re-run safely):

from hexarch_guardrails import Guardian
from hexarch_guardrails.exceptions import PolicyViolation

guardian = Guardian()

def request_delete(user_id: str) -> str:
    # 1) Persist intent to confirm (DB/cache)
    token = create_confirmation_token(user_id)
    send_approval_link(user_id, token)
    return token

def approve_delete(token: str) -> None:
    # 2) Mark token as approved in your DB
    mark_token_approved(token)

@guardian.check("safe_delete", context={"confirm": True})
def delete_user_data(user_id: str) -> None:
    db.delete(user_id)

def run_delete_if_approved(user_id: str, token: str) -> None:
    if not is_token_approved(token):
        raise PolicyViolation("Confirmation missing or expired")
    delete_user_data(user_id)

Key idea: store confirmation outside the SDK, then pass it in context.

B) Budget state in distributed environments

The demo uses in-memory counters (single-process only). For real systems, you need shared state.

Two common approaches:

1. OPA as a service + external state store

   • Run OPA as a sidecar/service.
   • Store budget usage in a database or cache (e.g., Postgres/Redis).
   • Your policy engine checks/updates that shared store during evaluation.

2. Custom decision service (recommended for multi-tenant systems)

   • Implement a small service that handles budgets centrally and returns decisions.

C) OPA dependency (what you should use in production)

The in-process evaluator used in the demo is not intended for production. For HA environments, deploy OPA as a sidecar or service and point the SDK to it:

guardian = Guardian(opa_url="http://opa:8181")

The SDK expects a decision response from the policy engine in the format:

{ "allowed": true|false, "reason": "...", "action": "warn|block" }

D) Custom decision service interface (standard contract)

If you want to swap OPA with your own service, the SDK only requires a client with:

class DecisionClient:
    def check_policy(self, policy_id: str, context: dict) -> dict:
        return {"allowed": True, "reason": "", "action": ""}

You can inject this by replacing the OPA client before creating Guardian (same technique used in the demo):

import hexarch_guardrails.guardian as guardian_module

guardian_module.OPAClient = DecisionClient  # use your own engine
guardian = Guardian()

E) “Upcoming” features vs core safety

Upcoming analytics (decision querying, metrics) do not affect enforcement. Core safety works today: if a policy blocks, a PolicyViolation is raised and your app can log or persist the denial.

For auditability now, log denials centrally:

try:
    run_sensitive_operation()
except PolicyViolation as exc:
    audit_log("policy_denied", reason=str(exc), context=...)

Features

• ✅ Zero-config - Auto-discovers hexarch.yaml
• ✅ Decorator-based - Drop in @guardian.check(policy_id)
• ✅ Policy-driven - YAML-based rules, no code changes
• ✅ Local-first - Works offline or with local OPA
• ✅ Pluggable - Works with any API/SDK
• ✅ Async-First - Full support for async def functions
• ✅ Fail-Safe - Denials don't crash your app (logged gracefully)
• ✅ Test-Friendly - Bypass policies in test environments
• ✅ Production-Ready - Used in live systems handling millions of requests

Examples

Budget Protection (OpenAI)

@guardian.check("api_budget")
def expensive_operation():
    # This call is protected by budget limits
    return openai.ChatCompletion.create(model="gpt-4", ...)

Rate Limiting

@guardian.check("rate_limit")
def send_discord_message(msg):
    return client.send(msg)

Safe File Operations

@guardian.check("safe_delete")
def delete_file(path):
    os.remove(path)

Prevent API Bill Shock (Async)

@guardian.check("openai_budget")
async def generate_content_batch(prompts: list[str]):
    return await asyncio.gather(*[
        openai.ChatCompletion.acreate(
            model="gpt-4",
            messages=[{"role": "user", "content": p}]
        ) for p in prompts
    ])

# hexarch.yaml enforces: max $100/month, warn at 80%

Business Hours Enforcement

@guardian.check("business_hours_only")
def send_customer_email_blast(emails: list[str]):
    for email in emails:
        send_marketing_email(email)

# Blocks execution outside 9am-5pm EST
# (Prevents accidental 3am customer notifications)

Discord/Slack Rate Limiting

@guardian.check("webhook_rate_limit")
def send_discord_alert(message: str):
    webhook.send(message)

# Automatically throttles to 5 msg/min (TOS compliant)

Database Migration Safety

@guardian.check("migration_safety")
def run_schema_migration(migration_file: str):
    db.execute_migration(migration_file)

# Requires: production env flag + confirmation + backup verified

Safe Automation Runner (CLI demo app)

A minimal, self-contained example that shows rate limits, budgets, and destructive safeguards enforced at the function boundary.

• Docs + code: examples/safe_automation_runner

---

Who Uses Hexarch?

• Solo Developers — Prevent accidental production disasters
• Hackathon Teams — Ship fast without breaking things
• Startups — Enforce compliance before you have a DevOps team
• AI/ML Engineers — Control GPU/API costs without code changes
• Enterprises — Policy-as-code for regulated industries (HIPAA, SOC 2)

"Saved us from a $3k Anthropic bill when an intern accidentally ran a batch job with Claude Opus instead of Haiku."
— Startup CTO using Hexarch in production

"We enforce 'no destructive operations after 4pm Friday' as policy. Game changer for weekend on-call."
— Platform Engineer

---

Why Not Just Use Try/Catch or Environment Variables?

Try/catch — Reactive. Runs code first, handles errors after damage is done.
Hexarch Guardrails — Proactive. Blocks execution before the function body runs.

Environment variables — Scattered across codebase, easy to bypass.
Hexarch Guardrails — Centralized policies in hexarch.yaml, enforced at the decorator boundary.

Manual checks in every function — Brittle, gets skipped during refactors.
Hexarch Guardrails — One decorator, policies evolve without touching code.

---

Documentation

• Policy Authoring Guide
• API Reference
• Examples

Admin CLI (v0.3.0+)

Hexarch includes a command-line interface for managing policies and monitoring decisions:

Installation

# Install with CLI extras
pip install hexarch-guardrails[cli]

Quick Start

# List all policies
hexarch-ctl policy list

# Export a policy
hexarch-ctl policy export ai_governance --format rego

# Validate policy syntax
hexarch-ctl policy validate ./policy.rego

# Compare policy versions
hexarch-ctl policy diff ai_governance

Available Commands

Policy Management:

• hexarch-ctl policy list - List all OPA policies
• hexarch-ctl policy export - Export policy to file or stdout
• hexarch-ctl policy validate - Validate OPA policy syntax
• hexarch-ctl policy diff - Compare policy versions

Upcoming (Phase 3-4):

• Decision querying and analysis
• Metrics and performance monitoring
• Configuration management

For detailed CLI documentation, see POLICY_COMMANDS_GUIDE.md

REST API Server (Phase 3)

Hexarch includes a hardened FastAPI server intended to back a UI/dev workflow.

Install server extras

pip install "hexarch-guardrails[server]"

Run locally (recommended)

hexarch-ctl serve api --host 127.0.0.1 --port 8099 --init-db --api-token dev-token

Notes:

• /health is public; most endpoints require a bearer token.
• API key management endpoints (/api-keys) are disabled by default and can be enabled explicitly with HEXARCH_API_KEY_ADMIN_ENABLED=true.

Credibility: OpenAPI fuzz scan (Schemathesis)

This repo includes a reproducible harness that starts the local FastAPI server with docs enabled and runs a Schemathesis scan against /openapi.json.

• Installs: pip install -e ".[server,credibility]"
• Runs (Windows): ./scripts/run_openapi_credibility_scan.ps1 -MaxExamples 25
• Output: evidence/credibility/openapi-schemathesis/<timestamp>/

The scan is configured with a conservative credibility check (not_a_server_error) to demonstrate resilience under generated inputs (no 5xx), without requiring that every auth error path is explicitly modeled in the OpenAPI spec.

Credibility: OWASP ZAP baseline scan (Docker)

This repo also includes an OWASP ZAP baseline scan harness (passive scan + spider) that produces HTML/JSON/XML/Markdown reports.

• Runs (Windows): ./scripts/run_zap_baseline_credibility_scan.ps1 -Mins 1 -MaxWaitMins 5
• Output: evidence/credibility/zap-baseline/<timestamp>/

Notes:

• By default it keeps auth enabled (hardened posture) and does not fail the command on warnings; it always writes the reports.
• For deeper crawling you can run with -AllowAnon (this changes server posture for the scan).
• To propagate ZAP exit codes (WARN/FAIL), pass -Strict.

Credibility: Policy correctness evals (golden cases)

Golden-case evaluation that exercises the decision engine via POST /authorize and writes a dated report.

• Cases file: evals/policy_cases.json (edit/extend this as you like)
• Runs (Windows): ./scripts/run_policy_credibility_evals.ps1
• Output: evidence/credibility/policy-evals/<timestamp>/ (report.md, results.json, server.log)

Smoke test (starts server, hits /health, stops)

PowerShell:

./scripts/smoke_api.ps1 -Port 8099

Bash:

./scripts/smoke_api.sh

n8n End-to-End (Single User Milestone)

For a complete local workflow that (1) calls /authorize, (2) calls a provider (Ollama), and (3) logs a tamper-evident provider-call event via /events/provider-calls, see:

• N8N_SINGLE_USER_MILESTONE.md

Node-RED End-to-End (Single User Milestone)

If you prefer an Apache-2.0 OSS orchestrator for guardrails testing (authorize → echo → log provider call), see:

• NODE_RED_SINGLE_USER_MILESTONE.md

License

MIT © Noir Stack LLC

See LICENSE for full details.

---

Publishing to PyPI (Maintainers Only)

Prerequisites

1. Install build tools:

   pip install --upgrade pip build twine
   

2. PyPI Account Setup:

   • Create account at https://pypi.org/account/register/
   • Enable 2FA (required for trusted publishers)
   • Generate API token: https://pypi.org/manage/account/token/
   • Save token securely (starts with pypi-)

3. Configure PyPI credentials:

   Option A: Token in .pypirc (recommended):

   # Create/edit ~/.pypirc (Linux/Mac) or %USERPROFILE%\.pypirc (Windows)
   [pypi]
   username = __token__
   password = pypi-YOUR_API_TOKEN_HERE
   

   Option B: Environment variable:

   # Linux/Mac
   export TWINE_USERNAME=__token__
   export TWINE_PASSWORD=pypi-YOUR_API_TOKEN_HERE
   
   # Windows PowerShell
   $env:TWINE_USERNAME = "__token__"
   $env:TWINE_PASSWORD = "pypi-YOUR_API_TOKEN_HERE"
   

Pre-Publish Checklist

• Update version in setup.py or pyproject.toml (use semantic versioning: 0.4.0, 1.0.0, etc.)
• Update CHANGELOG.md with release notes
• Ensure README.md renders correctly (PyPI uses this as long description)
• Run tests: pytest tests/
• Build locally to verify: python -m build
• Check package metadata: twine check dist/*
• Tag release in git: git tag v0.4.0 && git push origin v0.4.0

Publishing Steps

1. Navigate to package directory

cd hexarch-guardrails-py

2. Clean previous builds

# Linux/Mac
rm -rf build/ dist/ *.egg-info

# Windows PowerShell
Remove-Item -Recurse -Force build, dist, *.egg-info -ErrorAction SilentlyContinue

3. Build distribution packages

python -m build

This creates:

• dist/hexarch_guardrails-0.4.0-py3-none-any.whl (wheel - preferred)
• dist/hexarch-guardrails-0.4.0.tar.gz (source distribution)

4. Verify package contents

# Check metadata and README rendering
twine check dist/*

# Expected output:
# Checking dist/hexarch_guardrails-0.4.0-py3-none-any.whl: PASSED
# Checking dist/hexarch-guardrails-0.4.0.tar.gz: PASSED

5. Test upload to TestPyPI (optional but recommended)

# Upload to TestPyPI
twine upload --repository testpypi dist/*

# Test install from TestPyPI
pip install --index-url https://test.pypi.org/simple/ hexarch-guardrails==0.4.0

# Verify it works
python -c "from hexarch_guardrails import Guardian; print('✓ Package OK')"

6. Upload to production PyPI

twine upload dist/*

You'll see:

Uploading distributions to https://upload.pypi.org/legacy/
Uploading hexarch_guardrails-0.4.0-py3-none-any.whl
100% ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 25.2/25.2 kB • 00:00
Uploading hexarch-guardrails-0.4.0.tar.gz
100% ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 22.8/22.8 kB • 00:00

View at:
https://pypi.org/project/hexarch-guardrails/0.4.0/

7. Verify live package

# Wait 1-2 minutes for PyPI CDN propagation, then:
pip install --upgrade hexarch-guardrails

# Test
python -c "from hexarch_guardrails import Guardian; print(Guardian.__version__)"

8. Announce release

• Update GitHub release notes: https://github.com/no1rstack/Hexarch/releases
• Tweet/social media: "🚀 Hexarch Guardrails v0.4.0 now on PyPI: [link]"
• Update docs site if applicable

Troubleshooting

Error: 403 Forbidden during upload

• Check API token is valid and not expired
• Ensure you have maintainer/owner role on the package
• Verify .pypirc formatting (no spaces around =)

Error: File already exists

• PyPI does not allow re-uploading the same version
• Increment version number in setup.py/pyproject.toml
• Rebuild: python -m build

Error: Invalid distribution metadata

• Run twine check dist/* to see specific errors
• Common issues:
  • Missing README.md reference in setup.py
  • Invalid classifiers in setup.py
  • Non-UTF-8 characters in README.md

README not rendering on PyPI

• Ensure setup.py has long_description_content_type="text/markdown"
• Check for unsupported Markdown extensions (PyPI uses CommonMark)
• Test locally: python -m readme_renderer README.md -o /tmp/output.html

Package installs but imports fail

• Verify packages=find_packages() in setup.py includes all submodules
• Check MANIFEST.in includes necessary non-Python files
• Test in clean virtualenv: python -m venv test_env && source test_env/bin/activate

Automation (GitHub Actions)

For automated PyPI releases on git tags, see .github/workflows/publish-pypi.yml (if configured).

Example workflow trigger:

git tag v0.4.0
git push origin v0.4.0
# GitHub Actions automatically builds and publishes to PyPI

Security Best Practices

• Never commit .pypirc or API tokens to git
• Use PyPI API tokens (not password) — tokens are scoped and revocable
• Enable 2FA on PyPI account
• Use different tokens for TestPyPI vs production PyPI
• Rotate tokens quarterly or after team member departures
• Consider using GitHub's OIDC trusted publisher (no tokens needed)

Rollback Procedure

PyPI does not support deleting releases. If you need to rollback:

1. Yank the bad release (hides from pip install but keeps historical link):

   # Via PyPI web UI: Project → Releases → Manage → Yank
   

2. Publish fixed version:

   # Increment version: 0.4.0 → 0.4.1
   python -m build
   twine upload dist/*
   

3. Notify users:

   ⚠️ **Security Advisory**: hexarch-guardrails 0.4.0 has been yanked due to [issue].
   Please upgrade to 0.4.1 immediately:
   
   pip install --upgrade hexarch-guardrails
   

---

Questions? Open an issue: https://github.com/no1rstack/Hexarch/issues
Want to contribute? See CONTRIBUTING.md