AnomalyArmor Query Gateway

What is this?

This is the open-source SQL security layer that AnomalyArmor uses to control access to customer databases. Every query we execute against your database passes through this gateway.

Why open source? So you can verify exactly what queries we can and cannot run. No black boxes.

What does it do?

                      ┌─────────────────────────────────┐
                      │      AnomalyArmor Platform      │
                      └─────────────────────────────────┘
                                       │
                                       │
                                       ▼
┌────-─────────────────────────────────────────-────────────────────────-───────┐
│                          Query Security Gateway                               │
│   ┌─────────────────┐    ┌─────────────────-┐    ┌────────────────────────┐   │
│   │   SQL Parser    │ -> │ Access Validator │ -> │  Allow / Block + Log   │   │
│   │   (sqlglot)     │    │  (level rules)   │    │  (audit trail)         │   │
│   └─────────────────┘    └─────────────────-┘    └────────────────────────┘   │
└─────-─────────────────────────────────────────────────────────────────────────┘
                                        │
                                        | ALLOWED OPERATIONS ONLY
                                        ▼
                       ┌────────────────-────────────────┐
                       │         Your Database           │
                       └────────────────-────────────────┘

Validates SQL queries against three access levels:

The gateway parses your SQL using sqlglot and blocks queries that violate the configured access level. If parsing fails, the query is blocked (fail-closed).

How do we know it works?

97 tests covering access level enforcement, SQL parsing, and security edge cases.

See TEST_RESULTS.md for the full breakdown, or run them yourself:

pip install -e ".[dev]"
pytest -v

Overview

The Query Security Gateway provides a transparent, auditable layer for controlling what types of SQL queries can be executed against customer databases. It supports three access levels:

Installation

pip install anomalyarmor-query-gateway

Quick Start

from anomalyarmor_query_gateway import QuerySecurityGateway, AccessLevel

# Create gateway with desired access level
gateway = QuerySecurityGateway(
    access_level=AccessLevel.AGGREGATES,
    dialect="postgresql",
)

# Validate a query
result = gateway.validate_query_sync("SELECT COUNT(*) FROM users")

if result.allowed:
    print("Query is allowed")
else:
    print(f"Query blocked: {result.reason}")

Access Levels Explained

schema_only

Only allows queries against system/metadata tables:

• PostgreSQL: information_schema.*, pg_catalog.*
• MySQL: information_schema.*, mysql.*, performance_schema.*
• Databricks: information_schema.*, system.*
• ClickHouse: system.*
• SQLite: sqlite_master, sqlite_schema

aggregates

Allows aggregate functions but blocks raw column values:

# Allowed
"SELECT COUNT(*) FROM users"
"SELECT AVG(salary) FROM employees"
"SELECT MIN(created_at), MAX(created_at) FROM orders"

# Blocked
"SELECT * FROM users"
"SELECT email FROM users"
"SELECT salary FROM employees WHERE id = 1"

full

Allows any valid SELECT query.

Async Support

from anomalyarmor_query_gateway import QuerySecurityGateway, AccessLevel

gateway = QuerySecurityGateway(
    access_level=AccessLevel.AGGREGATES,
    dialect="postgresql",
)

# Async validation with audit logging
result = await gateway.validate_query(
    "SELECT COUNT(*) FROM users",
    metadata={"asset_id": "123", "user_id": "456"}
)

Audit Logging

Implement the AuditLoggerProtocol to log all query validation attempts:

from anomalyarmor_query_gateway import (
    QuerySecurityGateway,
    AccessLevel,
    AuditLoggerProtocol,
)

class MyAuditLogger(AuditLoggerProtocol):
    async def log_query(
        self,
        query: str,
        access_level: AccessLevel,
        dialect: str,
        allowed: bool,
        rejection_reason: str | None,
        metadata: dict,
    ) -> None:
        # Store to your audit log
        print(f"Query {'allowed' if allowed else 'blocked'}: {query[:50]}...")

gateway = QuerySecurityGateway(
    access_level=AccessLevel.AGGREGATES,
    dialect="postgresql",
    audit_logger=MyAuditLogger(),
)

Supported Dialects

• postgresql / postgres
• mysql
• databricks
• clickhouse
• sqlite

Security

This package is designed to be a security control layer. Key security features:

1. Fail-closed: If parsing fails, the query is blocked
2. Comment stripping: SQL comments are removed before parsing to prevent obfuscation
3. Recursive validation: Subqueries and CTEs are validated against the same rules
4. Window function blocking: Window functions blocked in aggregates mode (can expose row-level data)

Development

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run linter
ruff check src/ tests/

# Run type checker
mypy src/

License

Apache 2.0 - See LICENSE for details.

Contributing

See CONTRIBUTING.md for guidelines.

Security Issues

See SECURITY.md for reporting security vulnerabilities.