APILeak Developer Guide¶

# Example unit test
import pytest
from unittest.mock import Mock, patch
from core.config_manager import ConfigManager

class TestConfigManager:
    def setup_method(self):
        """Setup for each test method."""
        self.config_manager = ConfigManager()

    def test_load_valid_yaml_config(self):
        """Test loading a valid YAML configuration."""
        config_data = {
            'target': 'https://api.example.com',
            'modules': ['bola', 'auth']
        }

        with patch('builtins.open', mock_open(read_data=yaml.dump(config_data))):
            config = self.config_manager.load_config('test.yaml')

        assert config.target == 'https://api.example.com'
        assert 'bola' in config.modules

    def test_invalid_config_raises_validation_error(self):
        """Test that invalid config raises ValidationError."""
        invalid_config = {'invalid': 'config'}

        with patch('builtins.open', mock_open(read_data=yaml.dump(invalid_config))):
            with pytest.raises(ValidationError):
                self.config_manager.load_config('invalid.yaml')

# Example integration test
import pytest
import asyncio
from core.apileak_core import APILeakCore
from modules.owasp.bola_testing import BOLATestingModule

class TestBOLAIntegration:
    @pytest.mark.asyncio
    async def test_bola_testing_flow(self):
        """Test complete BOLA testing flow."""
        # Setup
        config = {
            'target': 'https://api.testserver.com',
            'auth': {'type': 'bearer', 'token': 'test-token'},
            'modules': {'bola': {'enabled': True}}
        }

        core = APILeakCore(config)
        bola_module = BOLATestingModule(core.http_engine)

        # Execute
        findings = await bola_module.run_tests()

        # Verify
        assert len(findings) > 0
        assert any(f.severity == 'CRITICAL' for f in findings)

from hypothesis import given, strategies as st
from core.response_analyzer import ResponseAnalyzer

class TestResponseAnalyzerProperties:

    @given(st.text())
    def test_sensitive_data_detection_never_false_positive_on_random_text(self, random_text):
        """Property: Random text should not be detected as sensitive data."""
        analyzer = ResponseAnalyzer()

        # Exclude actual sensitive patterns from random text
        assume(not any(pattern in random_text.lower() for pattern in 
                      ['password', 'api_key', 'secret', 'token']))

        result = analyzer.detect_sensitive_data(random_text)
        assert not result.has_sensitive_data

    @given(st.integers(min_value=200, max_value=599))
    def test_status_code_classification_consistency(self, status_code):
        """Property: Status code classification should be consistent."""
        analyzer = ResponseAnalyzer()

        classification1 = analyzer.classify_status_code(status_code)
        classification2 = analyzer.classify_status_code(status_code)

        assert classification1 == classification2

    @given(st.dictionaries(st.text(), st.text()))
    def test_header_analysis_preserves_original_headers(self, headers):
        """Property: Header analysis should not modify original headers."""
        analyzer = ResponseAnalyzer()
        original_headers = headers.copy()

        analyzer.analyze_security_headers(headers)

        assert headers == original_headers

# conftest.py
from hypothesis import settings, Verbosity

# Configure Hypothesis for CI/CD
settings.register_profile("ci", max_examples=1000, verbosity=Verbosity.verbose)
settings.register_profile("dev", max_examples=100, verbosity=Verbosity.normal)
settings.register_profile("debug", max_examples=10, verbosity=Verbosity.verbose)

# Load profile based on environment
import os
profile = os.getenv("HYPOTHESIS_PROFILE", "dev")
settings.load_profile(profile)

# modules/owasp/new_module.py
from typing import List, Dict, Any
from dataclasses import dataclass
from core.base_module import BaseTestingModule
from core.findings import Finding

@dataclass
class NewModuleConfig:
    """Configuration for the new OWASP module."""
    enabled: bool = True
    max_attempts: int = 100
    custom_payloads: List[str] = None

class NewOWASPModule(BaseTestingModule):
    """
    New OWASP testing module for [specific vulnerability].

    This module tests for [vulnerability description] by [testing approach].
    """

    def __init__(self, http_engine, config: NewModuleConfig):
        super().__init__(http_engine)
        self.config = config
        self.module_name = "new_owasp_module"

    async def run_tests(self) -> List[Finding]:
        """
        Execute the new OWASP tests.

        Returns:
            List of findings discovered during testing.
        """
        findings = []

        try:
            # Implement testing logic here
            findings.extend(await self._test_specific_vulnerability())

        except Exception as e:
            self.logger.error(f"Error in {self.module_name}: {e}")

        return findings

    async def _test_specific_vulnerability(self) -> List[Finding]:
        """Test for specific vulnerability patterns."""
        findings = []

        # Implementation details

        return findings

    def _create_finding(self, endpoint: str, details: Dict[str, Any]) -> Finding:
        """Create a standardized finding."""
        return Finding(
            module=self.module_name,
            endpoint=endpoint,
            severity=self._calculate_severity(details),
            title=f"[Vulnerability Name] detected",
            description=details.get('description', ''),
            evidence=details.get('evidence', {}),
            recommendation="Specific remediation steps"
        )

# core/module_registry.py
from modules.owasp.new_module import NewOWASPModule

OWASP_MODULES = {
    'bola': BOLATestingModule,
    'auth': AuthTestingModule,
    'property_auth': PropertyAuthModule,
    'function_auth': FunctionAuthModule,
    'resource': ResourceTestingModule,
    'ssrf': SSRFTestingModule,
    'new_module': NewOWASPModule,  # Add new module here
}

# tests/unit/test_new_module.py
import pytest
from unittest.mock import AsyncMock
from modules.owasp.new_module import NewOWASPModule, NewModuleConfig

class TestNewOWASPModule:
    def setup_method(self):
        self.http_engine = AsyncMock()
        self.config = NewModuleConfig()
        self.module = NewOWASPModule(self.http_engine, self.config)

    @pytest.mark.asyncio
    async def test_run_tests_returns_findings(self):
        """Test that run_tests returns a list of findings."""
        findings = await self.module.run_tests()
        assert isinstance(findings, list)

    @pytest.mark.asyncio
    async def test_vulnerability_detection(self):
        """Test specific vulnerability detection logic."""
        # Mock HTTP responses
        self.http_engine.request.return_value = MockResponse(
            status_code=200,
            text="vulnerable response"
        )

        findings = await self.module._test_specific_vulnerability()
        assert len(findings) > 0
        assert findings[0].severity in ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW']

# Good: Proper async/await usage
async def process_endpoints(self, endpoints: List[str]) -> List[Finding]:
    """Process multiple endpoints concurrently."""
    tasks = []
    semaphore = asyncio.Semaphore(10)  # Limit concurrent requests

    for endpoint in endpoints:
        task = self._process_single_endpoint(endpoint, semaphore)
        tasks.append(task)

    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r for r in results if isinstance(r, Finding)]

async def _process_single_endpoint(self, endpoint: str, semaphore: asyncio.Semaphore) -> Finding:
    """Process a single endpoint with rate limiting."""
    async with semaphore:
        response = await self.http_engine.request('GET', endpoint)
        return self._analyze_response(response)

# Bad: Blocking operations in async context
async def bad_example(self):
    time.sleep(1)  # Blocks the event loop
    requests.get(url)  # Synchronous HTTP request

# Good: Generator for large datasets
def generate_payloads(self) -> Iterator[str]:
    """Generate payloads on-demand to save memory."""
    for base_payload in self.base_payloads:
        for encoding in self.encodings:
            yield self._encode_payload(base_payload, encoding)

# Good: Context managers for resource cleanup
class HTTPEngine:
    async def __aenter__(self):
        self.session = httpx.AsyncClient()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.aclose()

# Usage
async with HTTPEngine() as engine:
    response = await engine.request('GET', url)

from functools import lru_cache
from typing import Dict, Any

class FrameworkDetector:
    def __init__(self):
        self._detection_cache: Dict[str, str] = {}

    @lru_cache(maxsize=1000)
    def detect_framework_by_headers(self, headers_tuple: tuple) -> str:
        """Cache framework detection results."""
        headers = dict(headers_tuple)
        return self._analyze_headers(headers)

    async def detect_framework(self, url: str) -> str:
        """Detect framework with caching."""
        if url in self._detection_cache:
            return self._detection_cache[url]

        response = await self.http_engine.request('GET', url)
        headers_tuple = tuple(sorted(response.headers.items()))

        framework = self.detect_framework_by_headers(headers_tuple)
        self._detection_cache[url] = framework

        return framework

# utils/logging_utils.py
import structlog
import logging
from typing import Any, Dict

def configure_logging(level: str = "INFO", format_json: bool = False) -> None:
    """Configure structured logging for APILeak."""

    processors = [
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
    ]

    if format_json:
        processors.append(structlog.processors.JSONRenderer())
    else:
        processors.append(structlog.dev.ConsoleRenderer())

    structlog.configure(
        processors=processors,
        wrapper_class=structlog.stdlib.BoundLogger,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

    logging.basicConfig(
        format="%(message)s",
        level=getattr(logging, level.upper()),
    )

# Usage in modules
class BOLATestingModule:
    def __init__(self, http_engine):
        self.logger = structlog.get_logger(__name__)

    async def run_tests(self):
        self.logger.info("Starting BOLA testing", module="bola", target=self.target)

        try:
            findings = await self._execute_tests()
            self.logger.info("BOLA testing completed", 
                           findings_count=len(findings),
                           critical_findings=len([f for f in findings if f.severity == 'CRITICAL']))
            return findings

        except Exception as e:
            self.logger.error("BOLA testing failed", 
                            error=str(e), 
                            exc_info=True)
            raise

# utils/profiling_utils.py
import cProfile
import pstats
import asyncio
from functools import wraps
from typing import Callable, Any

def profile_async(func: Callable) -> Callable:
    """Decorator to profile async functions."""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        profiler = cProfile.Profile()
        profiler.enable()

        try:
            result = await func(*args, **kwargs)
            return result
        finally:
            profiler.disable()
            stats = pstats.Stats(profiler)
            stats.sort_stats('cumulative')
            stats.print_stats(20)  # Top 20 functions

    return wrapper

# Usage
class APILeakCore:
    @profile_async
    async def run_all_tests(self):
        """Run all tests with profiling."""
        # Implementation
        pass

# Development dependency: memory-profiler
# pip install memory-profiler

from memory_profiler import profile

class ResponseAnalyzer:
    @profile
    def analyze_large_response(self, response_text: str):
        """Analyze response with memory profiling."""
        # Implementation that processes large responses
        pass

# Run with: python -m memory_profiler script.py

def analyze_response(self, response: httpx.Response, patterns: List[str]) -> AnalysisResult:
    """
    Analyze HTTP response for vulnerability patterns.

    Args:
        response: The HTTP response to analyze.
        patterns: List of regex patterns to search for.

    Returns:
        AnalysisResult containing findings and metadata.

    Raises:
        AnalysisError: If response analysis fails.

    Example:
        >>> analyzer = ResponseAnalyzer()
        >>> response = httpx.get("https://api.example.com")
        >>> result = analyzer.analyze_response(response, ["error", "exception"])
        >>> print(f"Found {len(result.findings)} issues")
    """
    # Implementation
    pass