OpenAPI Generator Security — Securing Generated Code and API Specs in CI/CD

import json
import re
from pathlib import Path

class SpecSecurityValidator:
    def __init__(self, spec_path):
        with open(spec_path) as f:
            self.spec = json.load(f)

    def validate(self):
        """Run all security checks on the spec."""
        issues = []
        issues.extend(self._check_missing_auth())
        issues.extend(self._check_injection_prone_params())
        issues.extend(self._check_unbounded_arrays())
        issues.extend(self._check_clear_text_schemes())
        issues.extend(self._check_excessive_scope())
        return issues

    def _check_missing_auth(self):
        """Find endpoints without authentication."""
        issues = []
        paths = self.spec.get('paths', {})
        security_global = self.spec.get('security', [])

        for path, methods in paths.items():
            for method, operation in methods.items():
                # Check per-operation security
                op_security = operation.get('security', security_global)
                if not op_security or op_security == [{}]:
                    issues.append({
                        'severity': 'HIGH',
                        'path': f"{method.upper()} {path}",
                        'issue': 'No authentication defined',
                        'fix': 'Add security requirement or global security'
                    })
        return issues

    def _check_injection_prone_params(self):
        """Find parameters vulnerable to injection."""
        issues = []
        for path, methods in self.spec.get('paths', {}).items():
            for method, operation in methods.items():
                params = operation.get('parameters', [])
                for param in params:
                    name = param.get('name', '')
                    schema = param.get('schema', {})
                    if any(kw in name.lower() for kw in
                           ['command', 'exec', 'eval', 'query', 'sql']):
                        issues.append({
                            'severity': 'CRITICAL',
                            'path': f"{method.upper()} {path}",
                            'param': name,
                            'issue': 'Parameter name suggests injection risk',
                            'fix': 'Add pattern validation or use enum'
                        })
        return issues

    def _check_excessive_scope(self):
        """Check for overly permissive security scopes."""
        issues = []
        for path, methods in self.spec.get('paths', {}).items():
            for method, operation in methods.items():
                security = operation.get('security', [])
                for sec_req in security:
                    for scheme, scopes in sec_req.items():
                        if 'admin' in scopes and method.upper() == 'GET':
                            issues.append({
                                'severity': 'MEDIUM',
                                'path': f"{method.upper()} {path}",
                                'issue': f"Read operation uses 'admin' scope",
                                'fix': 'Use more granular read scope'
                            })
        return issues

# Validate spec
validator = SpecSecurityValidator('openapi.yaml')
issues = validator.validate()
for issue in issues:
    print(f"[{issue['severity']}] {issue['path']}: {issue['issue']}")

# Custom template addition for Python client input validation

from pydantic import BaseModel, validator, Field
from typing import Optional, List
import re

class CreateThreatRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=255)
    severity: str = Field(...)
    indicator_type: str = Field(...)
    indicator_value: str = Field(...)
    description: Optional[str] = Field(None, max_length=5000)

    @validator('severity')
    def validate_severity(cls, v):
        allowed = ['low', 'medium', 'high', 'critical']
        if v.lower() not in allowed:
            raise ValueError(f'Severity must be one of: {allowed}')
        return v.lower()

    @validator('indicator_value')
    def validate_indicator(cls, v, values):
        indicator_type = values.get('indicator_type', '')

        if indicator_type == 'IP':
            if not re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', v):
                raise ValueError('Invalid IP address format')
        elif indicator_type == 'DOMAIN':
            if not re.match(r'^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', v):
                raise ValueError('Invalid domain format')
        elif indicator_type == 'HASH':
            if not re.match(r'^[a-fA-F0-9]{32,128}$', v):
                raise ValueError('Invalid hash format')

        return v

    @validator('description')
    def sanitize_description(cls, v):
        if v:
            # Strip HTML/script tags
            v = re.sub(r'<[^>]*>', '', v)
            return v.strip()
        return v

#!/usr/bin/env python3
"""
Secure CI/CD pipeline for OpenAPI generation.
"""
import subprocess
import hashlib
import json
from pathlib import Path

def verify_spec_integrity(spec_path, checksum_file):
    """Verify spec hasn't been tampered with."""
    current_hash = hashlib.sha256(Path(spec_path).read_bytes()).hexdigest()
    expected_hash = Path(checksum_file).read_text().strip()

    if current_hash != expected_hash:
        raise Exception(f"Spec integrity check failed for {spec_path}")
    print(f"Integrity check passed for {spec_path}")

def validate_spec_security(spec_path):
    """Run security validation on spec."""
    validator = SpecSecurityValidator(spec_path)
    issues = validator.validate()

    critical = [i for i in issues if i['severity'] == 'CRITICAL']
    high = [i for i in issues if i['severity'] == 'HIGH']

    if critical:
        raise Exception(
            f"CRITICAL security issues found: {len(critical)}. "
            "Pipeline blocked."
        )

    if high:
        print(f"WARNING: {len(high)} high-severity issues found")
        # May want to block here depending on policy

    print(f"Security validation passed: {len(issues)} issues found")

def scan_generated_code(output_dir):
    """Scan generated code for secrets and vulnerabilities."""
    # Check for hardcoded secrets
    result = subprocess.run([
        'gitleaks', 'detect',
        '--source', output_dir,
        '--no-git',
        '-v'
    ], capture_output=True, text=True)

    if result.returncode != 0:
        print("WARNING: Secrets detected in generated code:")
        print(result.stdout)

    # Check for vulnerable dependencies
    subprocess.run([
        'safety', 'check',
        '-r', f'{output_dir}/requirements.txt'
    ])

    # Run static analysis
    subprocess.run([
        'bandit', '-r', output_dir,
        '-ll'  # Only high confidence issues
    ])

def generate_securely(spec_path, generator, output_dir):
    """Generate code with security checks."""
    # Step 1: Verify integrity
    verify_spec_integrity(spec_path, 'spec.checksum')

    # Step 2: Validate spec security
    validate_spec_security(spec_path)

    # Step 3: Generate code
    subprocess.run([
        'openapi-generator-cli', 'generate',
        '-i', spec_path,
        '-g', generator,
        '-o', output_dir,
        '--additional-properties=performSecurityValidation=true'
    ], check=True)

    # Step 4: Scan generated code
    scan_generated_code(output_dir)

    print(f"Secure generation completed for {generator}")

# Run pipeline
generate_securely('openapi.yaml', 'python', '/tmp/secure-sdk')