# Create a virtual environment
python3 -m venv venv-security
source venv-security/bin/activate

# Install required libraries
# requests: for API calls, pydantic: for data validation
pip install requests pydantic python-dotenv

# Initialize project
mkdir security-scanner && cd security-scanner
npm init -y

# Install dependencies
# axios: HTTP client, zod: Schema validation, dotenv: Env management
npm install axios zod dotenv
npm install --save-dev typescript @types/node ts-node

import os
import requests
from typing import List, Dict
from pydantic import BaseModel, ValidationError
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Data Models for strict type checking
class Vulnerability(BaseModel):
    cve_id: str
    severity: str
    description: str
    affected_kernels: List[str]

class ScannerEngine:
    def __init__(self, api_key: str):
        self.api_key = api_key
        # Mock API endpoint - in production, use NIST NVD or Snyk API
        self.api_url = "https://api.security-provider.example/v1/check"

    def check_kernel_vulnerability(self, current_kernel: str) -> Dict:
        """
        Checks if the provided kernel version is vulnerable to VM Escape.
        """
        print(f"[*] Scanning kernel version: {current_kernel}...")
        
        try:
            # In a real scenario, we send the kernel version to the API
            # For this example, we simulate an API response
            response = self._simulate_api_call(current_kernel)
            
            # Validate response using Pydantic
            vulnerabilities = [Vulnerability(**v) for v in response]
            
            if vulnerabilities:
                return {"status": "VULNERABLE", "issues": vulnerabilities}
            return {"status": "SECURE", "issues": []}

        except ValidationError as e:
            return {"status": "ERROR", "message": f"Data integrity error: {e}"}
        except Exception as e:
            return {"status": "ERROR", "message": str(e)}

    def _simulate_api_call(self, kernel: str) -> List[Dict]:
        """Simulates a response from a vulnerability database."""
        # Critical VM Escape Vulnerability logic
        vulnerable_kernels = ["5.15.0-76", "5.10.0-21"]
        
        if kernel in vulnerable_kernels:
            return [{
                "cve_id": "CVE-2024-XXXXX",
                "severity": "CRITICAL",
                "description": "VM Escape via Intel/AMD hypervisor memory corruption.",
                "affected_kernels": vulnerable_kernels
            }]
        return []

if __name__ == "__main__":
    # Retrieve API key from environment
    API_KEY = os.getenv("SECURITY_API_KEY", "default_dev_key")
    
    scanner = ScannerEngine(API_KEY)
    
    # Test Case 1: Vulnerable Kernel
    result = scanner.check_kernel_vulnerability("5.15.0-76")
    print(f"Result: {result['status']}")
    if result['status'] == "VULNERABLE":
        for issue in result['issues']:
            print(f"  [!] {issue.cve_id} ({issue.severity}): {issue.description}")

    # Test Case 2: Secure Kernel
    result_secure = scanner.check_kernel_vulnerability("6.1.0")
    print(f"\nResult: {result_secure['status']}")

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

dotenv.config();

// Define the schema for a Vulnerability report
const VulnerabilitySchema = z.object({
  cve_id: z.string(),
  severity: z.enum(['LOW', 'MEDIUM', 'HIGH', 'CRITICAL']),
  description: z.string(),
  affected_kernels: z.array(z.string()),
});

type Vulnerability = z.infer<typeof VulnerabilitySchema>;

interface ScanResult {
  status: 'SECURE' | 'VULNERABLE' | 'ERROR';
  issues?: Vulnerability[];
  error?: string;
}

class SecurityScanner {
  private apiKey: string;

  constructor(apiKey: string) {
    this.apiKey = apiKey;
  }

  /**
   * Scans a kernel version against known vulnerabilities
   * @param kernelVersion The string representation of the Linux kernel
   */
  async scanKernel(kernelVersion: string): Promise<ScanResult> {
    try {
      console.log(`[i] Initiating scan for kernel: ${kernelVersion}`);

      // Simulating API call to a vulnerability database
      const responseData = await this.mockApiCall(kernelVersion);

      if (responseData.length === 0) {
        return { status: 'SECURE' };
      }

      // Validate all returned vulnerabilities against our schema
      const validatedIssues = responseData.map((issue) => 
        VulnerabilitySchema.parse(issue)
      );

      return {
        status: 'VULNERABLE',
        issues: validatedIssues,
      };
    } catch (error) {
      if (error instanceof z.ZodError) {
        return { status: 'ERROR', error: `Schema Validation Failed: ${error.message}` };
      }
      return { status: 'ERROR', error: (error as Error).message };
    }
  }

  private async mockApiCall(kernel: string): Promise<any[]> {
    // Simulated delay
    await new Promise((resolve) => setTimeout(resolve, 500));

    const vulnerableKernels = ['5.15.0-76', '5.10.0-21'];
    
    if (vulnerableKernels.includes(kernel)) {
      return [
        {
          cve_id: 'CVE-2024-9999',
          severity: 'CRITICAL',
          description: 'Hypervisor escape vulnerability detected on Intel/AMD.',
          affected_kernels: vulnerableKernels,
        },
      ];
    }
    return [];
  }
}

// Execution Logic
async function run() {
  const scanner = new SecurityScanner(process.env.SECURITY_API_KEY || 'dev_key');

  // Test Vulnerable
  const res1 = await scanner.scanKernel('5.15.0-76');
  console.log('Scan 1:', res1.status === 'VULNERABLE' ? '⚠️ VULNERABLE' : '✅ SECURE');
  if (res1.issues) res1.issues.forEach(i => console.log(`   - ${i.cve_id}: ${i.description}`));

  // Test Secure
  const res2 = await scanner.scanKernel('6.5.0');
  console.log('Scan 2:', res2.status === 'VULNERABLE' ? '⚠️ VULNERABLE' : '✅ SECURE');
}

run().catch(console.error);

# Security API Credentials
SECURITY_API_KEY=sk_live_51MzXy2L90_example_key

# Environment Settings
ENVIRONMENT=production
LOG_LEVEL=info

# Notification Webhooks (e.g., Slack or Microsoft Teams)
ALERTS_WEBHOOK_URL=https://hooks.slack.com/services/T000/B000/XXXX

# Pattern: Fail-Closed
try:
    result = scanner.check_kernel_vulnerability(ver)
except ConnectionError:
    # If we can't verify, we trigger an alert/block deployment
    trigger_emergency_alert("Scanner unreachable. Blocking deployment.")
    exit(1)

// Pattern: Alert-Only
if (result.status === 'VULNERABLE' && env === 'staging') {
    await sendSlackNotification(result.issues);
    console.warn("Vulnerability detected, but allowing build to continue in staging.");
}