Penetration Testing

Overview

Systematic security testing to identify, exploit, and document vulnerabilities in applications, networks, and infrastructure through simulated attacks.

When to Use

• Pre-production security validation
• Annual security assessments
• Compliance requirements (PCI-DSS, ISO 27001)
• Post-incident security review
• Third-party security audits
• Red team exercises

Implementation Examples

1. Automated Penetration Testing Framework

# pentest_framework.py
import requests
import socket
import subprocess
import json
from typing import List, Dict
from dataclasses import dataclass, asdict
from datetime import datetime

@dataclass
class Finding:
    severity: str
    category: str
    target: str
    vulnerability: str
    evidence: str
    remediation: str
    cvss_score: float

class PenetrationTester:
    def __init__(self, target: str):
        self.target = target
        self.findings: List[Finding] = []

    def test_sql_injection(self, url: str) -> None:
        """Test for SQL injection vulnerabilities"""
        print(f"Testing SQL injection on {url}")

        payloads = [
            "' OR '1'='1",
            "'; DROP TABLE users--",
            "' UNION SELECT NULL, NULL, NULL--",
            "1' AND 1=1--",
            "admin'--"
        ]

        for payload in payloads:
            try:
                response = requests.get(
                    url,
                    params={'id': payload},
                    timeout=5
                )

                # Check for SQL errors
                sql_errors = [
                    'mysql_fetch_array',
                    'SQLServer JDBC Driver',
                    'ORA-01756',
                    'PostgreSQL',
                    'sqlite3.OperationalError'
                ]

                for error in sql_errors:
                    if error in response.text:
                        self.findings.append(Finding(
                            severity='critical',
                            category='SQL Injection',
                            target=url,
                            vulnerability=f'SQL Injection detected with payload: {payload}',
                            evidence=f'Error message: {error}',
                            remediation='Use parameterized queries or prepared statements',
                            cvss_score=9.8
                        ))
                        break

            except Exception as e:
                print(f"Error testing {url}: {e}")

    def test_xss(self, url: str) -> None:
        """Test for Cross-Site Scripting vulnerabilities"""
        print(f"Testing XSS on {url}")

        payloads = [
            "<script>alert('XSS')</script>",
            "<img src=x onerror=alert('XSS')>",
            "javascript:alert('XSS')",
            "<svg onload=alert('XSS')>",
            "'-alert('XSS')-'"
        ]

        for payload in payloads:
            try:
                response = requests.get(
                    url,
                    params={'q': payload},
                    timeout=5
                )

                if payload in response.text:
                    self.findings.append(Finding(
                        severity='high',
                        category='Cross-Site Scripting',
                        target=url,
                        vulnerability=f'Reflected XSS detected with payload: {payload}',
                        evidence='Payload reflected in response without sanitization',
                        remediation='Implement output encoding and Content Security Policy',
                        cvss_score=7.3
                    ))
                    break

            except Exception as e:
                print(f"Error testing {url}: {e}")

    def test_authentication(self, login_url: str) -> None:
        """Test authentication mechanisms"""
        print(f"Testing authentication on {login_url}")

        # Test default credentials
        default_creds = [
            ('admin', 'admin'),
            ('admin', 'password'),
            ('root', 'root'),
            ('administrator', 'administrator')
        ]

        for username, password in default_creds:
            try:
                response = requests.post(
                    login_url,
                    data={'username': username, 'password': password},
                    timeout=5
                )

                if response.status_code == 200 and 'dashboard' in response.text.lower():
                    self.findings.append(Finding(
                        severity='critical',
                        category='Weak Authentication',
                        target=login_url,
                        vulnerability=f'Default credentials accepted: {username}/{password}',
                        evidence='Successful authentication with default credentials',
                        remediation='Enforce strong password policy and remove default accounts',
                        cvss_score=9.1
                    ))

            except Exception as e:
                print(f"Error testing credentials: {e}")

    def test_security_headers(self, url: str) -> None:
        """Test for missing security headers"""
        print(f"Testing security headers on {url}")

        try:
            response = requests.get(url, timeout=5)

            critical_headers = {
                'Strict-Transport-Security': 'HSTS not implemented',
                'X-Frame-Options': 'Clickjacking protection missing',
                'X-Content-Type-Options': 'MIME sniffing prevention missing',
                'Content-Security-Policy': 'CSP not implemented',
                'X-XSS-Protection': 'XSS protection header missing'
            }

            for header, description in critical_headers.items():
                if header not in response.headers:
                    self.findings.append(Finding(
                        severity='medium',
                        category='Missing Security Header',
                        target=url,
                        vulnerability=f'Missing header: {header}',
                        evidence=description,
                        remediation=f'Add {header} header to all responses',
                        cvss_score=5.3
                    ))

        except Exception as e:
            print(f"Error testing headers: {e}")

    def test_directory_traversal(self, url: str) -> None:
        """Test for directory traversal vulnerabilities"""
        print(f"Testing directory traversal on {url}")

        payloads = [
            '../../../etc/passwd',
            '..\\..\\..\\windows\\system32\\drivers\\etc\\hosts',
            '....//....//....//etc/passwd',
            '%2e%2e%2f%2e%2e%2f%2e%2e%2fetc%2fpasswd'
        ]

        for payload in payloads:
            try:
                response = requests.get(
                    url,
                    params={'file': payload},
                    timeout=5
                )

                # Check for Unix passwd file
                if 'root:' in response.text or 'daemon:' in response.text:
                    self.findings.append(Finding(
                        severity='critical',
                        category='Directory Traversal',
                        target=url,
                        vulnerability=f'Path traversal detected with payload: {payload}',
                        evidence='System file contents exposed',
                        remediation='Validate and sanitize file paths, use whitelist approach',
                        cvss_score=8.6
                    ))
                    break

            except Exception as e:
                print(f"Error testing traversal: {e}")

    def test_ssl_tls(self) -> None:
        """Test SSL/TLS configuration"""
        print(f"Testing SSL/TLS on {self.target}")

        try:
            result = subprocess.run(
                ['testssl.sh', '--jsonfile', 'ssl-results.json', self.target],
                capture_output=True,
                text=True,
                timeout=60
            )

            # Parse SSL test results
            # This is a simplified check
            weak_protocols = ['SSLv2', 'SSLv3', 'TLSv1.0']
            for protocol in weak_protocols:
                self.findings.append(Finding(
                    severity='high',
                    category='Weak SSL/TLS',
                    target=self.target,
                    vulnerability=f'Weak protocol enabled: {protocol}',
                    evidence='Outdated SSL/TLS protocol support',
                    remediation='Disable weak protocols, enforce TLS 1.2+',
                    cvss_score=7.5
                ))

        except Exception as e:
            print(f"SSL test error: {e}")

    def run_full_pentest(self, target_urls: List[str]) -> Dict:
        """Execute comprehensive penetration test"""
        for url in target_urls:
            self.test_sql_injection(url)
            self.test_xss(url)
            self.test_security_headers(url)
            self.test_directory_traversal(url)

        self.test_ssl_tls()

        return self.generate_report()

    def generate_report(self) -> Dict:
        """Generate comprehensive pentest report"""
        summary = {
            'critical': 0,
            'high': 0,
            'medium': 0,
            'low': 0
        }

        for finding in self.findings:
            if finding.severity in summary:
                summary[finding.severity] += 1

        report = {
            'timestamp': datetime.now().isoformat(),
            'target': self.target,
            'total_findings': len(self.findings),
            'summary': summary,
            'findings': [asdict(f) for f in self.findings],
            'risk_score': self._calculate_risk_score(),
            'recommendations': self._generate_recommendations()
        }

        with open('pentest-report.json', 'w') as f:
            json.dump(report, f, indent=2)

        return report

    def _calculate_risk_score(self) -> float:
        """Calculate overall risk score"""
        if not self.findings:
            return 0.0

        total_cvss = sum(f.cvss_score for f in self.findings)
        return round(total_cvss / len(self.findings), 2)

    def _generate_recommendations(self) -> List[str]:
        """Generate prioritized recommendations"""
        recommendations = []

        categories = {}
        for finding in self.findings:
            if finding.category not in categories:
                categories[finding.category] = []
            categories[finding.category].append(finding)

        for category, findings in sorted(
            categories.items(),
            key=lambda x: len(x[1]),
            reverse=True
        ):
            recommendations.append(
                f"Address {len(findings)} {category} vulnerabilities"
            )

        return recommendations[:5]

# Usage
if __name__ == '__main__':
    tester = PenetrationTester('https://example.com')

    target_urls = [
        'https://example.com/api/users',
        'https://example.com/search',
        'https://example.com/download'
    ]

    report = tester.run_full_pentest(target_urls)

    print("\n=== Penetration Test Report ===")
    print(f"Target: {report['target']}")
    print(f"Total Findings: {report['total_findings']}")
    print(f"Risk Score: {report['risk_score']}")
    print(f"\nFindings by Severity:")
    print(f"  Critical: {report['summary']['critical']}")
    print(f"  High: {report['summary']['high']}")
    print(f"  Medium: {report['summary']['medium']}")
    print(f"  Low: {report['summary']['low']}")

2. Burp Suite Automation Script

// burp-automation.js - Node.js Burp Suite integration
const axios = require('axios');
const fs = require('fs').promises;

class BurpSuiteAutomation {
  constructor(burpApiUrl = 'http://127.0.0.1:1337') {
    this.apiUrl = burpApiUrl;
    this.taskId = null;
  }

  async startScan(targetUrl) {
    console.log(`Starting Burp scan for ${targetUrl}`);

    const scanConfig = {
      urls: [targetUrl],
      scan_configurations: [
        {
          name: 'Crawl and Audit - Lightweight',
          type: 'NamedConfiguration'
        }
      ]
    };

    try {
      const response = await axios.post(
        `${this.apiUrl}/v0.1/scan`,
        scanConfig
      );

      this.taskId = response.data.task_id;
      console.log(`Scan started with task ID: ${this.taskId}`);

      return this.taskId;
    } catch (error) {
      console.error('Failed to start scan:', error.message);
      throw error;
    }
  }

  async getScanStatus() {
    if (!this.taskId) {
      throw new Error('No active scan task');
    }

    const response = await axios.get(
      `${this.apiUrl}/v0.1/scan/${this.taskId}`
    );

    return {
      taskId: this.taskId,
      status: response.data.scan_status,
      metrics: response.data.scan_metrics
    };
  }

  async waitForCompletion() {
    console.log('Waiting for scan to complete...');

    while (true) {
      const status = await this.getScanStatus();

      console.log(`Progress: ${status.metrics.crawl_requests_made} requests`);

      if (status.status === 'succeeded') {
        console.log('Scan completed successfully');
        break;
      } else if (status.status === 'failed') {
        throw new Error('Scan failed');
      }

      await new Promise(resolve => setTimeout(resolve, 10000));
    }
  }

  async getIssues() {
    if (!this.taskId) {
      throw new Error('No active scan task');
    }

    const response = await axios.get(
      `${this.apiUrl}/v0.1/scan/${this.taskId}/issues`
    );

    return response.data.issues;
  }

  async generateReport() {
    const issues = await this.getIssues();

    const report = {
      summary: {
        high: 0,
        medium: 0,
        low: 0,
        info: 0
      },
      issues: []
    };

    for (const issue of issues) {
      report.summary[issue.severity.toLowerCase()]++;

      report.issues.push({
        severity: issue.severity,
        confidence: issue.confidence,
        name: issue.name,
        path: issue.path,
        description: issue.description,
        remediation: issue.remediation
      });
    }

    await fs.writeFile(
      'burp-report.json',
      JSON.stringify(report, null, 2)
    );

    return report;
  }
}

// Usage
async function runBurpScan() {
  const burp = new BurpSuiteAutomation();

  await burp.startScan('https://example.com');
  await burp.waitForCompletion();

  const report = await burp.generateReport();

  console.log('\n=== Burp Suite Scan Results ===');
  console.log(`High: ${report.summary.high}`);
  console.log(`Medium: ${report.summary.medium}`);
  console.log(`Low: ${report.summary.low}`);
}

runBurpScan().catch(console.error);

Best Practices

✅ DO

• Get written authorization
• Define clear scope
• Use controlled environments
• Document all findings
• Follow responsible disclosure
• Provide remediation guidance
• Verify fixes after patching
• Maintain chain of custody

❌ DON'T

• Test production without approval
• Cause service disruption
• Exfiltrate sensitive data
• Share findings publicly
• Exceed authorized scope
• Use destructive payloads

Penetration Testing Phases

1. Reconnaissance: Information gathering
2. Scanning: Vulnerability identification
3. Exploitation: Proof of concept
4. Post-Exploitation: Assess impact
5. Reporting: Document findings
6. Remediation: Assist with fixes

Common Tools

• Burp Suite: Web application testing
• OWASP ZAP: Free security scanner
• Metasploit: Exploitation framework
• Nmap: Network scanning
• SQLMap: SQL injection testing
• testssl.sh: SSL/TLS testing
• Nikto: Web server scanner

Resources

• OWASP Testing Guide
• PTES Technical Guidelines
• OWASP ZAP Documentation
• Metasploit Unleashed