Metadata-Version: 2.4
Name: hexarch-guardrails
Version: 0.4.6b2
Summary: Lightweight policy-driven API protection and guardrails library
Home-page: https://www.noirstack.com/
Author: Noir Stack
Author-email: Noir Stack <hira@noirstack.com>
Maintainer: Hira
Maintainer-email: Hira <hira@noirstack.com>
License: MIT License
        
        Copyright (c) 2026 Noir Stack LLC
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Project-URL: Homepage, https://www.noirstack.com/
Project-URL: Repository, https://github.com/no1rstack/hexarch-guardrails
Project-URL: Documentation, https://github.com/no1rstack/hexarch-guardrails#readme
Project-URL: Issues, https://github.com/no1rstack/hexarch-guardrails/issues
Keywords: policy,guardrails,api,protection,opa
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.28.0
Requires-Dist: pyyaml>=6.0
Requires-Dist: python-dotenv>=0.21.0
Requires-Dist: click>=8.1.0
Requires-Dist: pydantic>=2.0
Requires-Dist: tabulate>=0.9.0
Requires-Dist: colorama>=0.4.6
Requires-Dist: sqlalchemy>=2.0.0
Requires-Dist: alembic>=1.13.0
Provides-Extra: cli
Requires-Dist: click>=8.1.0; extra == "cli"
Requires-Dist: pydantic>=2.0; extra == "cli"
Requires-Dist: tabulate>=0.9.0; extra == "cli"
Requires-Dist: colorama>=0.4.6; extra == "cli"
Provides-Extra: postgres
Requires-Dist: psycopg2-binary>=2.9.0; extra == "postgres"
Provides-Extra: server
Requires-Dist: fastapi>=0.110.0; extra == "server"
Requires-Dist: httpx<0.28,>=0.27.0; extra == "server"
Requires-Dist: uvicorn>=0.27.0; extra == "server"
Provides-Extra: credibility
Requires-Dist: schemathesis>=3.30.0; extra == "credibility"
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0; extra == "dev"
Requires-Dist: black>=23.0; extra == "dev"
Requires-Dist: flake8>=5.0; extra == "dev"
Dynamic: author
Dynamic: home-page
Dynamic: license-file
Dynamic: maintainer
Dynamic: requires-python

# 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?**

```python
# 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?**

```python
# 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:**

```python
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**](https://github.com/no1rstack/hexarch-guardrails/tree/main/examples/safe_automation_runner)

---

## Source-of-truth

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

## Installation

```bash
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`)

```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

```python
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):

```python
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.

```python
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:

```python
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:

```python
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:

```python
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:

```python
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/](examples/confirmation_patterns/) for runnable demos of all three patterns. Run:

```bash
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

```python
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)**
```python
from hexarch_guardrails.audit import SQLiteAuditStore
store = SQLiteAuditStore("audit.db")  # File-based, single-machine
```

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

**Null Store (Backward Compatibility)**
```python
from hexarch_guardrails.audit import NullAuditStore
store = NullAuditStore()  # No-op, zero overhead
```

### Query API Examples

**Get Decision History**
```python
# 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**
```python
# 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**
```python
# 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)**
```python
# All decisions for a specific user (for compliance queries)
user_trail = audit_logger.get_user_audit_trail(
    user_id="alice@example.com",
    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**
```python
# 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**
```bash
hexarch-ctl audit list --limit 50
hexarch-ctl audit list --policy-id sensitive_delete --limit 20
```

**Show Blocked Operations**
```bash
hexarch-ctl audit blocked --timeframe 24h
hexarch-ctl audit blocked --timeframe 7d --format json
```

**Policy Statistics**
```bash
hexarch-ctl audit stats --policy-id api_budget
hexarch-ctl audit stats --policy-id sensitive_delete --format table
```

**User Audit Trail**
```bash
hexarch-ctl audit trail --user-id alice@example.com
hexarch-ctl audit trail --user-id bob --limit 100 --format json
```

**Export to CSV/JSON**
```bash
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**
```bash
hexarch-ctl audit cleanup --days 90 --dry-run  # Preview deletions
hexarch-ctl audit cleanup --days 90             # Actually delete
```

### CLI Configuration

**SQLite (Default)**
```bash
export HEXARCH_AUDIT_DB=/path/to/decisions.db
hexarch-ctl audit list
```

**PostgreSQL**
```bash
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) | `"alice@example.com"` |
| `duration_ms` | Policy evaluation time (ms) | `15` |
| `metadata` | Custom tags (optional JSON) | `{"env": "prod", "version": "1.2"}` |

### Production Patterns

**Pattern A: Centralized Audit Database**
```python
# 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**
```python
@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="alice@example.com", data={"key": "value"})
```

**Pattern C: Custom Metadata for Filtering**
```python
# 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)**
```python
#!/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("security@company.com", "Daily Guardrails Report", body)
```

### Complete Working Example

See [examples/audit_example.py](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

```bash
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):

```python
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:

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

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

```json
{ "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:

```python
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):

```python
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:

```python
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)

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

### Rate Limiting

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

### Safe File Operations

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

### Prevent API Bill Shock (Async)

```python
@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

```python
@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

```python
@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

```python
@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](https://github.com/no1rstack/hexarch-guardrails/tree/main/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](https://github.com/no1rstack/hexarch-guardrails/blob/main/docs/POLICY_GUIDE.md)
- [API Reference](https://github.com/no1rstack/hexarch-guardrails/blob/main/docs/API_REFERENCE.md)
- [Examples](https://github.com/no1rstack/hexarch-guardrails/tree/main/examples)

## Admin CLI (v0.3.0+)

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

### Installation

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

### Quick Start

```bash
# 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](https://github.com/no1rstack/hexarch-guardrails/blob/main/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:

```bash
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 (admin@hexarch.local / admin)
```

Then configure your application:

```python
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**](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:

```bash
# 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:

```python
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](https://github.com/no1rstack/hexarch-guardrails/blob/main/hexarch_guardrails/rules.py)

---

### 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](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)
```bash
docker-compose -f docker-compose-embedded.yml up -d
```

**2. Sidecar Pattern** (shared OPA)
```bash
docker-compose -f docker-compose-sidecar.yml up -d
```

**3. Main Pattern** (production baseline)
```bash
docker-compose up -d
```

---

### Database Migrations with Alembic

Safely evolve your schema over time:

```bash
# 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](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):

```bash
# 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](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)**
```python
# 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)**
```python
# 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](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

```bash
pip install "hexarch-guardrails[server]"
```

### Run locally (recommended)

```bash
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:

```powershell
./scripts/smoke_api.ps1 -Port 8099
```

Bash:

```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](https://github.com/no1rstack/hexarch-guardrails/blob/main/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](https://github.com/no1rstack/hexarch-guardrails/blob/main/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**:
   ```bash
   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):
   ```bash
   # Create/edit ~/.pypirc (Linux/Mac) or %USERPROFILE%\.pypirc (Windows)
   [pypi]
   username = __token__
   password = pypi-YOUR_API_TOKEN_HERE
   ```

   **Option B: Environment variable**:
   ```bash
   # 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

```bash
cd hexarch-guardrails-py
```

#### 2. Clean previous builds

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

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

#### 3. Build distribution packages

```bash
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

```bash
# 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)

```bash
# 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

```bash
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

```bash
# 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:
```bash
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):
   ```bash
   # Via PyPI web UI: Project → Releases → Manage → Yank
   ```

2. **Publish fixed version**:
   ```bash
   # Increment version: 0.4.0 → 0.4.1
   python -m build
   twine upload dist/*
   ```

3. **Notify users**:
   ```markdown
   ⚠️ **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](https://github.com/no1rstack/Hexarch/blob/main/CONTRIBUTING.md)
