hexarch-guardrails 0.4.6b2

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

---

Audit System: Track Every Decision

Guardrails includes a built-in audit system to log every policy decision for compliance, troubleshooting, and analytics. All decisions (allowed, blocked, warned) are captured with full context and queryable via Python API or CLI.

Why Audit Logging?

Compliance: Prove which operations were blocked and why for SOC2/ISO27001 audits
Troubleshooting: Debug production policy failures without diving into raw logs
Analytics: Track block rates, identify problematic users/policies, find patterns
Governance: Generate audit reports for security reviews and incident response

Quick Start: Enable Audit Logging

from hexarch_guardrails import Guardian
from hexarch_guardrails.audit import DecisionLogger, SQLiteAuditStore

# Initialize audit storage
audit_logger = DecisionLogger(SQLiteAuditStore("decisions.db"))

# Pass to Guardian - all decisions are now logged automatically
guardian = Guardian(policy_file="hexarch.yaml", audit_logger=audit_logger)

@guardian.check("sensitive_delete")
def delete_records(record_ids):
    db.delete_many(record_ids)

# Every call is logged with timestamp, policy, decision, reason, context
delete_records([1, 2, 3])

Storage Backends

SQLite (Development)

from hexarch_guardrails.audit import SQLiteAuditStore
store = SQLiteAuditStore("audit.db")  # File-based, single-machine

PostgreSQL (Production)

from hexarch_guardrails.audit import PostgresAuditStore
store = PostgresAuditStore(
    host="db.example.com",
    database="hexarch",
    user="audit_user",
    password="..."
)

Null Store (Backward Compatibility)

from hexarch_guardrails.audit import NullAuditStore
store = NullAuditStore()  # No-op, zero overhead

Query API Examples

Get Decision History

# All decisions for a specific policy
decisions = audit_logger.get_decision_history(
    policy_id="sensitive_delete",
    limit=100,
    offset=0
)

for d in decisions:
    print(f"{d.timestamp}: {d.decision} - {d.reason}")
    print(f"  Function: {d.function_name}")
    print(f"  User: {d.user_id}")
    print(f"  Duration: {d.duration_ms}ms")

Find Blocked Operations

# All blocks in last 24 hours
blocked = audit_logger.get_blocked_operations(timeframe="24h", limit=50)

# Available timeframes: "24h", "7d", "30d"
weekly_blocks = audit_logger.get_blocked_operations(timeframe="7d")

Policy Statistics

# Get block rate and top reasons for a policy
stats = audit_logger.get_policy_stats("sensitive_delete")

print(f"Total decisions: {stats['total_decisions']}")
print(f"Block rate: {stats['block_rate']}%")
print(f"Top block reasons:")
for reason in stats['top_block_reasons']:
    print(f"  - {reason['reason']}: {reason['count']} times")

User Audit Trail (Compliance)

# All decisions for a specific user (for compliance queries)
user_trail = audit_logger.get_user_audit_trail(
    user_id="[e-mail]",
    limit=100
)

# Generate CSV report for auditors
import csv
with open("user_audit.csv", "w") as f:
    writer = csv.writer(f)
    writer.writerow(["Timestamp", "Policy", "Decision", "Reason"])
    for d in user_trail:
        writer.writerow([d.timestamp, d.policy_id, d.decision, d.reason])

Cleanup Old Records

# Delete audit records older than 90 days
deleted_count = audit_logger.cleanup_old_records(days=90)
print(f"Deleted {deleted_count} old records")

CLI Commands

Query audit logs directly from the command line:

List Recent Decisions

hexarch-ctl audit list --limit 50
hexarch-ctl audit list --policy-id sensitive_delete --limit 20

Show Blocked Operations

hexarch-ctl audit blocked --timeframe 24h
hexarch-ctl audit blocked --timeframe 7d --format json

Policy Statistics

hexarch-ctl audit stats --policy-id api_budget
hexarch-ctl audit stats --policy-id sensitive_delete --format table

User Audit Trail

hexarch-ctl audit trail --user-id [e-mail]
hexarch-ctl audit trail --user-id bob --limit 100 --format json

Export to CSV/JSON

hexarch-ctl audit export --output audit_report.csv --format csv
hexarch-ctl audit export --output audit_data.json --format json --policy-id api_budget

Cleanup Old Records

hexarch-ctl audit cleanup --days 90 --dry-run  # Preview deletions
hexarch-ctl audit cleanup --days 90             # Actually delete

CLI Configuration

SQLite (Default)

export HEXARCH_AUDIT_DB=/path/to/decisions.db
hexarch-ctl audit list

PostgreSQL

export PGPASSWORD=your_password
hexarch-ctl audit list --pg-host db.example.com --pg-database hexarch --pg-user audit_user

Logged Fields

Every audit decision includes:

Field	Description	Example
decision_id	Unique UUID	"550e8400-e29b-41d4-a716..."
timestamp	UTC timestamp	2026-02-02T18:30:45Z
policy_id	Policy that was evaluated	"sensitive_delete"
function_name	Guarded function	"delete_records"
decision	Outcome enum	"allowed", "blocked", "warned"
reason	Why decision was made	"Daily budget exceeded"
context	Full evaluation context (JSON)	{"user": "alice", "count": 150}
user_id	User who triggered (optional)	"[e-mail]"
duration_ms	Policy evaluation time (ms)	15
metadata	Custom tags (optional JSON)	{"env": "prod", "version": "1.2"}

Production Patterns

Pattern A: Centralized Audit Database

# All services log to shared PostgreSQL
from hexarch_guardrails.audit import PostgresAuditStore, DecisionLogger

store = PostgresAuditStore(
    host="audit-db.internal",
    database="centralized_audit",
    user="service_account",
    password=os.getenv("AUDIT_DB_PASSWORD")
)

audit_logger = DecisionLogger(store)
guardian = Guardian(audit_logger=audit_logger)

Pattern B: Add User Context to Every Decision

@guardian.check("api_call")
def call_external_api(user_id, data):
    # Guardian automatically logs user_id from decorator kwargs
    return api.post(data)

# User context is captured automatically
call_external_api(user_id="[e-mail]", data={"key": "value"})

Pattern C: Custom Metadata for Filtering

# Add custom metadata to decisions
@guardian.check("deployment", metadata={"environment": "production", "team": "backend"})
def deploy_service():
    k8s.deploy()

# Later, query by metadata (requires custom query implementation)
prod_decisions = audit_logger.get_decision_history(policy_id="deployment")
backend_decisions = [d for d in prod_decisions if d.metadata.get("team") == "backend"]

Pattern D: Daily Audit Reports (Cron)

#!/usr/bin/env python3
# daily_audit_report.py - Run via cron

from hexarch_guardrails.audit import DecisionLogger, PostgresAuditStore
import smtplib

audit = DecisionLogger(PostgresAuditStore(...))

# Get yesterday's blocked operations
blocked = audit.get_blocked_operations(timeframe="24h")

# Email security team
body = f"Blocked operations last 24h: {len(blocked)}\n\n"
for d in blocked[:10]:  # Top 10
    body += f"- {d.timestamp}: {d.policy_id} blocked {d.function_name} ({d.reason})\n"

send_email("[e-mail]", "Daily Guardrails Report", body)

Complete Working Example

See examples/audit_example.py for a runnable demonstration showing:

• Enabling audit logging
• Making guarded calls (allowed and blocked)
• Querying decision history
• Generating statistics
• Exporting user audit trails

cd examples
python audit_example.py

---

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

---

Production Deployment

Managing your own infrastructure? We've provided complete deployment automation to remove friction.

One-Command Local Setup

Deploy PostgreSQL + OPA policy engine locally:

docker-compose up -d

# Services started:
# - PostgreSQL (port 5432) - Audit database with schema auto-initialized
# - OPA (port 8181) - Policy engine with example policies loaded
# - pgAdmin (port 5050) - Database management UI ([e-mail] / admin)

Then configure your application:

from hexarch_guardrails import Guardian
from hexarch_guardrails.audit import DecisionLogger, PostgresAuditStore

audit_store = PostgresAuditStore(
    host="localhost",
    database="hexarch_audit",
    user="hexarch_user",
    password="hexarch_secure_password_change_me"  # Change in docker-compose.yml!
)

guardian = Guardian(
    policy_file="hexarch.yaml",
    opa_url="http://localhost:8181",
    audit_logger=DecisionLogger(store=audit_store)
)

Production Architectures

We provide infrastructure-as-code for multiple deployment patterns:

Architecture	Setup Time	Monthly Cost	Best For
Docker Compose (included)	2 min	$0	Development & testing
Kubernetes (YAML templates)	30 min	$100-500	Enterprise, high-availability
AWS (RDS + Lambda/ECS) (guide)	20 min	~$40	Serverless, low-ops
Google Cloud (guide)	20 min	~$35	Auto-scaling, GCP-native

Complete Infrastructure Guide

See INFRASTRUCTURE_SETUP.md for:

✅ Docker Compose - One file, three services (postgres + OPA + pgAdmin)
✅ SQL Schema - Auto-initialized with indexes and views
✅ OPA Policies - 3 example production policies (safe_delete, api_budget, rate_limit)
✅ Kubernetes - StatefulSet templates, ConfigMaps, resource limits
✅ Cloud Providers - Step-by-step AWS RDS, Google Cloud SQL, Azure guides
✅ Security - TLS setup, role-based access, network isolation
✅ Monitoring - Health check utility, Prometheus metrics, alerting patterns
✅ Backup/Recovery - Automated daily backups, restore procedures
✅ Troubleshooting - Common issues and solutions

Health Checking

Monitor your deployment health:

# Local health check
python healthcheck.py

# Output:
# ✓ PostgreSQL            OK - 1,247 total decisions (142 blocked, 89% allowed)
# ✓ OPA Service           OK
# ✓ OPA Policies         OK - 3 policies loaded (safe_delete, api_budget, rate_limit)

# Remote/production health check
python healthcheck.py \
  --postgres-host db.prod.company.com \
  --opa-url http://opa.prod.company.com:8181

# Show audit statistics
python healthcheck.py --summary

Responsibility Model

We provide:

• Docker Compose for local development (one command)
• SQL schema with indexes and views
• OPA policy examples
• Kubernetes templates
• Cloud provider guides (not managed services, but clear procedures)
• Health monitoring utilities
• Backup & disaster recovery scripts

You maintain:

• Database operations (backups, scaling, PITR)
• OPA policy updates and testing
• Infrastructure monitoring and alerting
• Network security and access control
• Secrets rotation and key management
• Compliance and audit procedures

This is transparent, vendor-neutral infrastructure — no lock-in, standard tools, full control.

---

Phase 2: SimpleRulesEngine & Advanced Deployment (0.4.6+)

SimpleRulesEngine: Lightweight Policy Without OPA

For users who don't need OPA's full power, use embedded SimpleRulesEngine:

from hexarch_guardrails.rules import RulesEngine, Rule, Condition

# Define simple rules
rules = [
    Rule(
        id="block_deletes",
        name="Require approval for deletes",
        conditions=[Condition("operation", "equals", "delete")],
        actions=["require_approval"],
        priority=10,
    ),
    Rule(
        id="rate_limit",
        name="Limit to 100 req/min",
        conditions=[Condition("requests_per_minute", "greater_than", 100)],
        actions=["block"],
        priority=5,
    )
]

# Create engine
engine = RulesEngine(rules=rules)

# Evaluate in your code
context = {"operation": "delete", "user_id": "alice"}
result = engine.evaluate(context)

if result.blocked:
    raise PermissionError(result.reason)

Benefits:

• ✅ No OPA dependency needed
• ✅ Sub-millisecond latency
• ✅ Embedded in your application
• ✅ Full Python type support
• ✅ Easy to test and debug

Best for: Single-service apps, development, simple policies

See: API Reference

---

Deployment Patterns

Choose the right architecture for your needs:

Pattern	Latency	Complexity	Cost	Scale
Embedded (SimpleRulesEngine)	<1ms	⭐ Very Low	$0-50/mo	~10K/sec
Sidecar (OPA per pod)	2-5ms	⭐⭐ Low	$250-2200/mo	~50K/sec
Central Service (Shared OPA)	10-50ms	⭐⭐⭐ Medium	$1500-10K/mo	500K+/sec
Service Mesh (Istio)	3-8ms	⭐⭐⭐⭐ High	$600-3500/mo	100K+/sec

See: DEPLOYMENT_PATTERNS.md for detailed architectures with Kubernetes YAML, cloud provider guides, cost analysis, and pro/con comparisons.

---

Docker Compose Templates

We provide multiple docker-compose configurations:

1. Embedded Pattern (smallest footprint)

docker-compose -f docker-compose-embedded.yml up -d

2. Sidecar Pattern (shared OPA)

docker-compose -f docker-compose-sidecar.yml up -d

3. Main Pattern (production baseline)

docker-compose up -d

---

Database Migrations with Alembic

Safely evolve your schema over time:

# Check current version
alembic current

# Apply latest migrations
alembic upgrade head

# See migration history
alembic history

# Rollback if needed
alembic downgrade -1

See: SCHEMA_MIGRATION.md for:

• Zero-downtime migration procedures
• Common scenarios (add field, add index, change type)
• Testing strategies
• Rollback procedures
• Performance optimization

---

Data Migration: 0.4.5 → 0.4.6+

If upgrading from 0.4.5 (no audit system) to 0.4.6+ (with audit):

# Show migration plan
python migration-guide.py --plan

# Test migration without changes
python migration-guide.py --dry-run

# Execute migration
python migration-guide.py --execute

# Rollback instructions
python migration-guide.py --rollback

See: migration-guide.py for complete upgrade workflow with backup/restore procedures.

---

Audit Data Retention Policy

Keep audit logs aligned with compliance and costs:

Time-based (simple)

# Keep 90 days
audit_store.cleanup_old_records(
    before_date=datetime.now() - timedelta(days=90)
)

Tiered (recommended)

• Last 30 days: PostgreSQL hot storage ($0.04/GB/mo)
• 31-90 days: S3 Glacier Instant ($0.004/GB/mo)
• 91+ days: S3 Glacier Deep Archive ($0.001/GB/mo)

By severity (risk-based)

# Keep blocks 1 year, routine logs 30 days
cleanup_by_severity = {
    "BLOCK": timedelta(days=365),
    "WARN": timedelta(days=90),
    "ALLOW": timedelta(days=30),
}

See: AUDIT_RETENTION_POLICY.md for:

• Compliance requirements (GDPR, HIPAA, SOX, PCI-DSS)
• Storage cost analysis
• Retention strategies by org size
• Cleanup script examples
• Kubernetes CronJob templates
• Compliance archival procedures

---

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