# Create a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows use `venv\Scripts\activate`

# Install necessary libraries
# sqlalchemy: For secure query construction
# requests: For communicating with Metabase API
# pydantic: For strict data validation
pip install sqlalchemy requests pydantic python-dotenv

# Initialize project
npm init -y

# Install dependencies
# axios: For API requests
# zod: For schema validation (essential for preventing injection)
# dotenv: For environment management
npm install axios zod dotenv
npm install --save-dev typescript @types/node

import os
import requests
from typing import Dict, Any
from pydantic import BaseModel, Field, validator
from dotenv import load_dotenv

load_dotenv()

# 1. Define a strict schema for incoming query requests
class SecureQueryRequest(BaseModel):
    question_id: int = Field(..., gt=0)  # Must be a positive integer
    parameters: Dict[str, Any] = Field(default_factory=dict)
    
    @validator('parameters')
    def prevent_sql_keywords(cls, v):
        """Custom validator to block common SQL injection patterns in params."""
        forbidden = ["DROP", "DELETE", "UNION", "--", ";"]
        for key, value in v.items():
            if isinstance(value, str):
                if any(keyword in value.upper() for keyword in forbidden):
                    raise ValueError(f"Potential SQL Injection detected in parameter: {key}")
        return v

class MetabaseSecurityProxy:
    def __init__(self):
        self.base_url = os.getenv("METABASE_URL")
        self.api_key = os.getenv("METABASE_API_KEY")
        self.session = requests.Session()
        self.session.headers.update({"X-Metabase-Session": self.api_key})

    def execute_safe_query(self, request_data: Dict[str, Any]):
        try:
            # 2. Validate input against schema before it ever touches the API
            validated_data = SecureQueryRequest(**request_data)
            
            # 3. Construct the API endpoint (Metabase API structure)
            endpoint = f"{self.base_url}/api/card/{validated_data.question_id}/query"
            
            # 4. Execute request
            response = self.session.post(endpoint, json=validated_data.parameters)
            response.raise_for_status()
            
            return response.json()

        except ValueError as ve:
            print(f"[SECURITY ALERT] Validation Failed: {ve}")
            return {"error": "Invalid request parameters", "code": 400}
        except requests.exceptions.RequestException as re:
            print(f"[ERROR] API Connection Issue: {re}")
            return {"error": "Internal service error", "code": 500}

# --- Example Usage ---
if __name__ == "__main__":
    proxy = MetabaseSecurityProxy()

    # Scenario A: Valid Request
    print("Running valid request...")
    print(proxy.execute_safe_query({"question_id": 10, "parameters": {"user_id": 123}}))

    # Scenario B: Malicious Injection Attempt
    print("\nRunning malicious request...")
    print(proxy.execute_safe_query({
        "question_id": 10, 
        "parameters": {"user_id": "123; DROP TABLE users;"}
    }))

import axios, { AxiosError } from 'axios';
import { z } from 'zod';
import * as dotenv from 'dotenv';

dotenv.config();

// 1. Define strict schema for query parameters
const QueryParamSchema = z.object({
  questionId: z.number().positive(),
  filters: z.record(z.union([z.string(), z.number()])).refine((params) => {
    // Prevent common SQL injection characters in filter values
    const sqlRegex = /(--|;|DROP|UNION|DELETE|UPDATE|SELECT)/i;
    return !Object.values(params).some(val => typeof val === 'string' && sqlRegex.test(val));
  }, {
    message: "Potential SQL injection detected in filter values"
  })
});

type QueryParam = z.infer<typeof QueryParamSchema>;

class MetabaseClient {
  private readonly baseUrl: string;
  private readonly sessionToken: string;

  constructor() {
    this.baseUrl = process.env.METABASE_URL || '';
    this.sessionToken = process.env.METABASE_API_KEY || '';
  }

  async safeQuery(input: unknown): Promise<any> {
    try {
      // 2. Validate input
      const validatedData = QueryParamSchema.parse(input);

      // 3. Execute request via Axios
      const response = await axios.post(
        `${this.baseUrl}/api/card/${validatedData.questionId}/query`,
        validatedData.filters,
        {
          headers: { 'X-Metabase-Session': this.sessionToken }
        }
      );

      return response.data;
    } catch (error) {
      if (error instanceof z.ZodError) {
        console.error('[SECURITY ALERT] Input Validation Error:', error.errors);
        throw new Error('Invalid input parameters');
      } else if (error instanceof AxiosError) {
        console.error('[API ERROR] Metabase communication failed:', error.message);
        throw new Error('Upstream service error');
      }
      throw error;
    }
  }
}

// --- Example Usage ---
async function runDemo() {
  const client = new MetabaseClient();

  // Test Case 1: Success
  try {
    console.log('Attempting safe query...');
    const data = await client.safeQuery({ questionId: 5, filters: { category: 'electronics' } });
    console.log('Success:', data);
  } catch (e: any) {
    console.error(e.message);
  }

  // Test Case 2: Injection Attempt
  try {
    console.log('\nAttempting malicious query...');
    await client.safeQuery({ 
      questionId: 5, 
      filters: { category: "electronics'; DROP TABLE users;--" } 
    });
  } catch (e: any) {
    console.error('Caught expected error:', e.message);
  }
}

runDemo();

# The base URL of your Metabase instance
METABASE_URL=https://your-metabase-instance.com

# The Session Token (Obtained via /api/session endpoint)
# DO NOT use admin credentials for application-level API calls.
# Use a dedicated service account with minimal permissions.
METABASE_API_KEY=your_secure_session_token_here