Skip to content

Security: middesurya/metalquery

Security

SECURITY.md

MetalQuery NLP2SQL Industrial Security Layer

Comprehensive Defense Against Prompt Injection, Flipping, & Red Teaming Status: Production-Ready | Framework: Django 3.8+ / FastAPI | Database: PostgreSQL 12+ Compliance: IEC 62443 SL-2/SL-3

Executive Summary

This document provides a complete industrial-grade security implementation for NLP2SQL systems that defends against:

Threat Reference Defense
Prompt Injection Attacks OWASP LLM01:2025 PromptSignatureValidator
Prompt Flipping Jailbreaks FlipAttack - 78.97% ASR FlippingDetector
Red Team Attacks 6 vulnerability categories RedTeamDetector
SQL Injection OWASP A03:2021 SQLInjectionValidator
Unauthorized Access IEC 62443 RBAC + Token Auth

Core Security Principle

┌─────────────────────────────────────────────────────────────────────────────────┐
│                    LLM NEVER TOUCHES DATABASE DIRECTLY                          │
│                                                                                  │
│   ┌─────────────┐         ┌─────────────┐         ┌─────────────┐              │
│   │     LLM     │         │   DJANGO    │         │  DATABASE   │              │
│   │   (Groq)    │         │   (8000)    │         │ (PostgreSQL)│              │
│   └──────┬──────┘         └──────┬──────┘         └──────┬──────┘              │
│          │                       │                       │                      │
│          │ Generates             │ Validates             │                      │
│          │ SQL TEXT              │ & Executes            │                      │
│          │ only                  │                       │                      │
│          ▼                       ▼                       │                      │
│   ┌─────────────────────────────────────────────────────┐│                      │
│   │              Django is the GATEKEEPER               ││                      │
│   │  • Validates all SQL before execution               │◄─────────────────────┘│
│   │  • Enforces RBAC permissions                        │                       │
│   │  • Owns database connection                         │                       │
│   │  • LLM has NO credentials, NO connection            │                       │
│   └─────────────────────────────────────────────────────┘                       │
└─────────────────────────────────────────────────────────────────────────────────┘

12-Layer Security Architecture

USER INPUT (Untrusted)
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 1: Rate Limiting (Django)                                  │
│          30 requests/min per IP                                  │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 2: Token Validation (Django)                               │
│          Bearer token → users_usertoken → users_user             │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 3: RBAC Service (Django)                                   │
│          Token → User → Roles → Permissions → allowed_tables     │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 4: Flipping Detector (NLP Service - Port 8003)             │
│          Detects jailbreak attempts (4 flipping modes)           │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 5: Prompt Validator (NLP Service)                          │
│          Injection signature detection (15+ patterns)            │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 6: Red Team Detector (NLP Service)                         │
│          Attack pattern recognition (6 categories)               │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 7: Guardrails AI (NLP Service)                             │
│          PII and profanity filtering                             │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 8: Query Guard (NLP Service)                               │
│          Relevance and intent verification                       │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 9: Schema Filter (NLP Service)                             │
│          LLM only sees user's allowed_tables                     │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼ (LLM generates SQL TEXT - no DB access)
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 10: SQL Validator (Django)                                 │
│           Block DROP, DELETE, INSERT, UPDATE, GRANT              │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 11: Table Validator (Django - Defense in Depth)            │
│           Verify SQL tables ⊆ allowed_tables                     │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────────────────────────────────┐
│ LAYER 12: Audit Logger (Django)                                  │
│           Log all queries with user context                      │
└─────────────────────────────────────────────────────────────────┘
    │
    ▼
DATABASE (Only Django executes SQL)

RBAC Implementation

Token-Based Authentication Flow

Authorization: Bearer <token>
        │
        ▼
┌─────────────────────────────────────────────────────────────────┐
│ STEP 1: Token Lookup                                             │
│         users_usertoken.key = token → user_id                    │
└─────────────────────────────────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────────────────────────────────┐
│ STEP 2: User Lookup                                              │
│         users_user.id = user_id                                  │
│         Check: is_superuser?                                     │
│         → If YES: Return ALL 29 tables                           │
│         → If NO: Continue to role lookup                         │
└─────────────────────────────────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────────────────────────────────┐
│ STEP 3: Role Lookup                                              │
│         users_userrole → role_id                                 │
└─────────────────────────────────────────────────────────────────┘
        │
        ├───────────────────────────────────────┐
        ▼                                       ▼
┌─────────────────────────┐         ┌─────────────────────────┐
│ STEP 4a: Function Codes │         │ STEP 4b: KPI Metrics    │
│ users_rolepermission    │         │ users_role_kpis         │
│ (where view=true)       │         │                         │
│                         │         │                         │
│ PLT_CFG → plant_plant   │         │ OEE → kpi_oee           │
│ FUR_MNT → furnace_*     │         │ YIELD → kpi_yield       │
│ LOG_BOOK → log_book_*   │         │ DOWNTIME → kpi_downtime │
│ TAP_PROC → tap_*        │         │ (21 total mappings)     │
└────────────┬────────────┘         └────────────┬────────────┘
             │                                   │
             └───────────────┬───────────────────┘
                             ▼
┌─────────────────────────────────────────────────────────────────┐
│ STEP 5: Build allowed_tables                                     │
│         {"kpi_oee", "plant_plant", "furnace_furnaceconfig", ...} │
└─────────────────────────────────────────────────────────────────┘

RBAC Service Implementation

File: backend/chatbot/services/rbac_service.py

class RBACService:
    def get_allowed_tables(self, token: str) -> tuple[list[str], Optional[str]]:
        """
        Resolve token to allowed tables.

        Returns:
            (allowed_tables, error_message)
        """
        # 1. Validate token
        user = self._get_user_from_token(token)
        if not user:
            return [], "Invalid or expired token"

        # 2. Superuser gets all tables
        if user.is_superuser:
            return list(ALL_EXPOSED_TABLES), None

        # 3. Get user's roles
        roles = self._get_user_roles(user)

        # 4. Build allowed tables from permissions
        allowed = set()
        for role in roles:
            # Function code permissions
            for perm in role.permissions.filter(view=True):
                tables = FUNCTION_TABLE_MAPPING.get(perm.function_code, [])
                allowed.update(tables)

            # KPI metric permissions
            for kpi in role.kpi_metrics.all():
                tables = KPI_TABLE_MAPPING.get(kpi.code, [])
                allowed.update(tables)

        return list(allowed), None

Permission Mappings

File: backend/ignis/schema/exposed_tables.py

# Function code → Tables
FUNCTION_TABLE_MAPPING = {
    "PLT_CFG": ["plant_plant"],
    "FUR_MNT": ["furnace_furnaceconfig", "furnace_config_parameters"],
    "LOG_BOOK": ["log_book_furnace_down_time_event", "log_book_reasons",
                 "log_book_downtime_type_master"],
    "TAP_PROC": ["core_process_tap_production", "core_process_tap_process",
                 "core_process_tap_grading"],
}

# KPI metric → Tables
KPI_TABLE_MAPPING = {
    "OEE": ["kpi_overall_equipment_efficiency_data"],
    "YIELD": ["kpi_yield_data"],
    "DOWNTIME": ["kpi_downtime_data"],
    "MTBF": ["kpi_mean_time_between_failures_data"],
    "MTTR": ["kpi_mean_time_to_repair_data"],
    "DEFECT": ["kpi_defect_rate_data"],
    "ENERGY_EFF": ["kpi_energy_efficiency_data"],
    "ENERGY_USE": ["kpi_energy_used_data"],
    # ... 21 total mappings
}

Database Tables for RBAC

Table Purpose
users_usertoken Token → user_id mapping
users_user User details, is_superuser flag
users_userrole User → role assignments
users_rolepermission Role → function code permissions
users_role_kpis Role → KPI metric permissions

Layer 1: NLP Service Security

1.1 Prompt Flipping Detection

File: nlp_service/security/flipping_detector.py

Detects 4 flipping modes based on FlipAttack research (ICLR 2025):

Mode Example Attack Detection Method
Word Order "bomb a build to how" Reverse word sequence harmful check
Char in Word "woH ot dliub" Per-word reversal detection
Char in Sentence "bmob a dliub ot woH" Full sentence reversal
Fool Model Conflicting instructions Task mismatch detection

1.2 Prompt Signature Validation

File: nlp_service/security/flipping_detector.py

Validates prompts against 15+ known attack signatures:

Attack Type Pattern Example Severity
Direct Injection "ignore previous instructions" 0.8
Role Assumption "system prompt", "admin mode" 0.7
Code Execution "execute command", "run code" 0.9
SQL Injection "UNION SELECT", "DROP TABLE" 0.95
Data Exfiltration "leak", "extract", "dump" 0.8

Layer 2: Schema Filtering

File: nlp_service/prompts_v2.py

The NLP service filters the schema before sending to LLM:

def build_prompt_with_schema(question: str, allowed_tables: list = None):
    """Build prompt with filtered schema based on RBAC permissions."""

    if allowed_tables:
        # Filter TABLE_SCHEMA to only include allowed tables
        filtered_schema = {
            table: info
            for table, info in TABLE_SCHEMA.items()
            if table in allowed_tables
        }
    else:
        filtered_schema = TABLE_SCHEMA

    # LLM only sees user's permitted tables
    schema_text = format_schema(filtered_schema)
    return f"{SYSTEM_PROMPT}\n\nAvailable tables:\n{schema_text}"

Layer 3: Defense in Depth (Django)

SQL Validation

File: backend/chatbot/views.py

Django validates SQL BEFORE execution:

def validate_and_execute_sql(sql: str, allowed_tables: list):
    """Validate SQL against RBAC permissions before execution."""

    # 1. Block dangerous keywords
    dangerous = ['DROP', 'DELETE', 'INSERT', 'UPDATE', 'GRANT', 'TRUNCATE']
    sql_upper = sql.upper()
    for keyword in dangerous:
        if keyword in sql_upper:
            return None, f"Dangerous SQL keyword: {keyword}"

    # 2. Extract tables from SQL
    tables_in_sql = extract_tables_from_sql(sql)

    # 3. Verify tables are in allowed_tables
    unauthorized = tables_in_sql - set(allowed_tables)
    if unauthorized:
        return None, f"Access denied to tables: {unauthorized}"

    # 4. Execute validated SQL
    return execute_sql(sql), None

SQL Comment Stripping (NLP Service)

File: nlp_service/sql_guardrails.py

Before validation, SQL comments are stripped to:

  • Allow LLM to add helpful comments without triggering security blocks
  • Prevent comment-based injection attacks
# Remove SQL comments (single-line -- and multi-line /* */)
sql_no_comments = re.sub(r'--[^\n]*', '', sql)  # Remove -- comments
sql_no_comments = re.sub(r'/\*.*?\*/', '', sql_no_comments, flags=re.DOTALL)

Error Responses

Scenario HTTP Code Response
No token 401 {"error": "Authentication required"}
Invalid token 401 {"error": "Invalid or expired token"}
No permissions 403 {"error": "No table access permissions"}
Unauthorized table 403 {"error": "Access denied to tables: xxx"}
SQL injection 200 Blocked by validator
Jailbreak attempt 200 Blocked by flipping detector

Security Test Results

RBAC Tests (16/16 Passing)

Test Status
No token → 401 ✅
Invalid token → 401 ✅
Expired token → 401 ✅
No permissions → 403 ✅
Superuser → 29 tables ✅
Limited user → subset ✅
Table not in allowed → 403 ✅
Defense in depth ✅

SQL Injection Tests (43/43 Blocked)

Category Tests Status
DROP statements 5 ✅ Blocked
DELETE statements 5 ✅ Blocked
UNION attacks 8 ✅ Blocked
Stacked queries 10 ✅ Blocked
Comment injection 8 ✅ Blocked
Encoding bypass 7 ✅ Blocked

Red Team Tests

Category Attacks Expected Block
Prompt Injection 6 100%
Flipping 4 100%
Reward Hacking 4 95%
Data Exfiltration 5 100%
Deceptive Alignment 4 90%
Privilege Escalation 4 100%

File Structure

poc_nlp_tosql/
├── backend/                          # Django Backend (Port 8000)
│   ├── chatbot/
│   │   ├── views.py                 # RBAC enforcement, SQL validation
│   │   └── services/
│   │       └── rbac_service.py      # Token → allowed_tables
│   └── ignis/
│       └── schema/
│           └── exposed_tables.py    # Permission mappings
│
└── nlp_service/                      # FastAPI NLP Service (Port 8003)
    └── security/
        ├── __init__.py              # Module exports
        ├── flipping_detector.py     # FlippingDetector, PromptSignatureValidator
        ├── sql_validator.py         # SQLInjectionValidator
        ├── anomaly_detector.py      # AnomalyDetector, RedTeamDetector
        └── audit_logger.py          # AuditLogger

Running Security Tests

# RBAC tests
cd backend
python test_rbac_defense.py

# SQL injection tests
python test_sql_injection.py

IEC 62443 Compliance Matrix

Requirement Implementation Status
SL-2: Access Control Token-based RBAC + table-level permissions ✅
SL-2: Audit Logging Comprehensive audit with user context ✅
SL-2: Rate Limiting 30 req/min per IP ✅
SL-2: SQL Injection Prevention 12-layer validation ✅
SL-3: Anomaly Detection Behavioral analysis engine ✅
SL-3: Defense in Depth Django re-validates NLP output ✅
SL-3: Threat Assessment Red team simulation ✅

Key Metrics

Metric Target Actual
Attack Block Rate >90% 97%
False Positive Rate <10% <5%
Query Overhead <100ms 35-50ms
Audit Trail Completeness 100% 100%
RBAC Tests Passing 100% 100% (16/16)
SQL Injection Tests 100% 100% (43/43)

Conclusion

This implementation provides 12-layer defense with:

  • ✅ Token-based authentication (Django owns all DB access)
  • ✅ Role-based table-level access control (RBAC)
  • ✅ Prompt injection attacks blocked (direct & indirect)
  • ✅ Prompt flipping jailbreaks blocked
  • ✅ Red team attacks blocked (6 categories)
  • ✅ SQL injection attacks blocked (43 test cases)
  • ✅ Defense-in-depth (Django validates NLP output)
  • ✅ Schema filtering (LLM only sees allowed tables)
  • ✅ Comprehensive audit logging

Core Principle: LLM generates SQL text only. Django is the single gatekeeper that validates and executes all queries.

Compliance: IEC 62443 SL-2/SL-3 Ready

Last Updated: 2026-01-09

There aren’t any published security advisories