⚡ Key Features

• AST Query Validator: In-memory SQL AST query analyzer detecting tautologies (WHERE 1=1), destructive DDL (DROP TABLE, TRUNCATE), unindexed queries, and row read ceilings.
• Dynamic PII Redactor: Format-preserving pseudonymization and regex redaction for SSNs, Luhn-valid credit card numbers, email addresses, and JWT/API tokens.
• Two-Phase Commit Distributed Sandbox: Dry-run simulation with automatic rollback and single-use HMAC-SHA256 commit tokens.
• Transparent Middleware: DB-API 2.0 cursor wrappers (DBShieldConnectionWrapper) and SQLAlchemy hooks (attach_dbshield) with zero application refactoring.
• AI Agent Framework Integrations:
  • LangChain: DBShieldSQLDatabase and DBShieldQueryTool
  • CrewAI: DBShieldDatabaseTool
  • LlamaIndex: DBShieldLlamaIndexTransform
• 100% Air-Gapped & Zero Telemetry: Operates strictly local-first within air-gapped private VPC networks.

📦 Installation

pip install dbshield-defense

Note: In code, import as standard import dbshield or from dbshield import DBShieldGuard.

🚀 Usage Guide

1. In-Memory AST SQL Guard

Validate queries generated by LLMs before execution:

from dbshield import DBShieldGuard

guard = DBShieldGuard()

# 1. Block destructive operations
result = guard.validate_query("DROP TABLE users")
print(result.allowed)     # False
print(result.risk_level)  # "CRITICAL"
print(result.violations)  # [Violation(code='DESTRUCTIVE_DDL_PROHIBITED', ...)]

# 2. Block unconstrained mutations and tautologies
result = guard.validate_query("DELETE FROM orders WHERE 1=1")
print(result.allowed)     # False

2. Dynamic PII Redaction & Format Preservation

from dbshield import PiiRedactor

redactor = PiiRedactor()

text = "User SSN is 000-12-3456 and email is john@acme.com"
sanitized = redactor.redact_text(text)
print(sanitized)
# -> "User SSN is [REDACTED:SSN:9705bb] and email is j***n@a****e.com"

3. Transparent SQLAlchemy Integration

from sqlalchemy import create_engine
from dbshield import attach_dbshield

engine = create_engine("postgresql://user:pass@localhost:5432/proddb")

# Attach DBShield interceptor hook
attach_dbshield(engine)

# All queries executed via engine are now validated and PII-sanitized automatically!

4. LangChain Integration

from dbshield.integrations.langchain import DBShieldSQLDatabase, DBShieldQueryTool
from langchain_community.utilities import SQLDatabase

raw_db = SQLDatabase.from_uri("sqlite:///analytics.db")
shield_db = DBShieldSQLDatabase(raw_db)

# Create an agent tool that refuses destructive DDL or tautology queries
tool = DBShieldQueryTool(db=shield_db)

5. CrewAI Database Tool

from dbshield.integrations.crewai import DBShieldDatabaseTool
import sqlite3

conn = sqlite3.connect("company.db")
db_tool = DBShieldDatabaseTool(connection=conn)

# When an autonomous CrewAI agent generates a query:
output = db_tool._run("SELECT name, email FROM customers WHERE active = 1")
print(output)  # Rows returned with sensitive PII automatically masked!

6. LlamaIndex Transform

from dbshield.integrations.llamaindex import DBShieldLlamaIndexTransform

transform = DBShieldLlamaIndexTransform()
safe_response = transform.postprocess_response("Query returned user SSN 000-12-3456")

🔒 Security & Compliance

• Zero Cloud Exfiltration: Runs 100% locally with DBSHIELD_AIR_GAPPED=1.
• Pre-filled SIG Lite / CAIQ: Download the Vendor Risk Assessment Packet at ritualdev.in/trust.
• Statutory Alignment: Designed for India DPDP Act 2023, EU GDPR, and SOC 2 Type II.

📄 License

MIT © RitualDev Studio