Automated Penetration Testing Framework

Enterprise-Grade Security Assessment Platform

A comprehensive, automated penetration testing framework designed for security professionals to conduct authorized security assessments. Built as part of a university research project (TM470), this framework orchestrates industry-standard security tools with intelligent finding correlation and detailed reporting.

⚠️ EDUCATIONAL USE ONLY: This tool is designed for authorized security testing in controlled lab environments. Unauthorized use against systems you don't own or have explicit permission to test is illegal.

🎯 Project Overview

This framework was developed to demonstrate automated penetration testing methodologies following industry standards (NIST, OWASP, CEH v12). It combines multiple security assessment phases with intelligent finding aggregation and compliance mapping.

Key Principle: Zero-dependency vulnerability detection with built-in intelligence - no external APIs or third-party scanning services required for baseline configuration assessment.

✨ Features

Core Capabilities

• Multi-Phase Penetration Testing: Reconnaissance, scanning, enumeration, exploitation, persistence, and post-exploitation
• Automated Tool Orchestration: Integrates Nmap, Nikto, SQLMap, and custom vulnerability scanners
• Baseline Configuration Analysis: Built-in detection for end-of-life systems, firewall status, missing antivirus, and dangerous protocol exposure
• Real-time Progress Tracking: WebSocket-based live updates during assessments
• Comprehensive Reporting: Export findings in PDF, CSV, and JSON formats
• Compliance Mapping: Automatic mapping to NIST CSF, ISO 27001, PCI DSS, HIPAA
• CVSS 3.1 Scoring: Industry-standard vulnerability severity ratings

Enterprise Features

• Role-Based Access Control (RBAC): Multi-user support with granular permissions
• Audit Logging: Complete audit trail for compliance and forensics
• API Rate Limiting: Protection against abuse
• JWT Authentication: Secure token-based authentication
• Health Monitoring: Comprehensive system health checks
• Caching Layer: Redis-backed caching for performance

📁 Architecture

TM470-F1746464/
├── api/                         # REST API layer
│   ├── auth/                   # Authentication & authorization
│   ├── export/                 # Report generation (PDF/CSV/JSON)
│   ├── health/                 # Health checks & monitoring
│   ├── middleware/             # Rate limiting, CORS, validation
│   └── websocket/              # Real-time communication
│
├── core/                        # Core business logic
│   ├── orchestrator/           # Assessment orchestration
│   │   ├── enhanced_orchestrator.py     # Main orchestrator
│   │   └── finding_aggregator.py        # Finding consolidation
│   ├── scanning/               # Vulnerability scanning
│   │   ├── enhanced_vulnerability_scanner.py  # Service-level scanner
│   │   └── vulnerability_scanning/      # Protocol-specific checks
│   ├── vulnerability/          # Baseline vulnerability detection
│   │   └── baseline_checker.py          # Configuration assessment
│   ├── exploitation/           # Exploitation modules
│   ├── persistence/            # Persistence mechanisms
│   ├── post_exploitation/      # Post-exploitation activities
│   ├── audit/                  # Audit logging
│   ├── cache/                  # Caching layer
│   └── tools/                  # Security tool integrations
│
├── database/                    # Database models & migrations
│   ├── models.py               # SQLAlchemy models
│   └── migrations/             # Alembic migrations
│
├── frontend/                    # React-based web interface
│   ├── src/
│   │   ├── components/         # React components
│   │   ├── pages/              # Page layouts
│   │   └── services/           # API integration
│   └── public/                 # Static assets
│
├── deployment/                  # Deployment configurations
│   ├── windows/                # Windows-specific deployment
│   └── linux/                  # Linux deployment scripts
│
├── docs/                        # Documentation
│   ├── BASELINE_VULNERABILITY_CHECKER.md  # Baseline checker guide
│   ├── API.md                  # API documentation
│   └── DEVELOPMENT.md          # Development guide
│
├── tests/                       # Test suite
│   ├── test_baseline_integration.py     # Baseline checker tests
│   ├── test_orchestrator.py    # Orchestrator tests
│   └── test_api.py             # API endpoint tests
│
├── main.py                      # Application entry point
├── cli.py                       # Command-line interface
├── requirements.txt             # Python dependencies
└── README.md                    # This file

🚀 Quick Start

Prerequisites

System Requirements:

• OS: Windows Server 2016+ / Windows 10+ (Primary) or Ubuntu 20.04+ (Alternative)
• Python: 3.11 or higher
• Database: PostgreSQL 15+
• Cache: Redis 7+
• RAM: Minimum 8GB (16GB recommended)
• Storage: 20GB free space

Required Tools (for full functionality):

• Nmap 7.80+
• Nikto 2.1+
• SQLMap 1.5+
• Metasploit Framework (optional)

Installation

Windows (Recommended Platform)

# Clone repository
git clone https://github.com/zy538324/TM470-F1746464.git
cd TM470-F1746464

# Create virtual environment
python -m venv venv
.\venv\Scripts\Activate.ps1

# Install dependencies
pip install -r requirements.txt

# Configure environment
copy .env.example .env
# Edit .env with your database credentials

# Initialize database
python cli.py init-db

# Create admin user
python cli.py create-admin --username admin --email [email protected]

# Start application
python main.py

Linux (Ubuntu/Debian)

# Clone repository
git clone https://github.com/zy538324/TM470-F1746464.git
cd TM470-F1746464

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

# Install dependencies
pip install -r requirements.txt

# Configure environment
cp .env.example .env
# Edit .env with your database credentials

# Initialize database
python cli.py init-db

# Create admin user
python cli.py create-admin --username admin --email [email protected]

# Start application
python main.py

Configuration

Create a .env file in the project root:

# Application
FLASK_ENV=production
FLASK_DEBUG=False
SECRET_KEY=your-cryptographically-secure-secret-key-here

# Database
DATABASE_URL=postgresql://pentester:secure_password@localhost:5432/pentester_db

# Redis
REDIS_URL=redis://localhost:6379/0

# JWT
JWT_SECRET_KEY=your-jwt-secret-key-here
JWT_ACCESS_TOKEN_EXPIRES=3600
JWT_REFRESH_TOKEN_EXPIRES=2592000

# Celery
CELERY_BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/0

# Security
API_RATE_LIMIT=100/hour
CORS_ORIGINS=http://localhost:3000,http://localhost:5000

📖 Usage

Starting the Application

Development Mode:

.\venv\Scripts\Activate.ps1
python main.py

Production Mode (Windows Services):

Start-Service PentestFramework
Start-Service PentestFramework-Celery
Start-Service PentestFramework-Beat

Production Mode (Linux systemd):

sudo systemctl start pentester
sudo systemctl start pentester-celery
sudo systemctl start pentester-beat

Running an Assessment

Via Web Interface

1. Navigate to http://localhost:5000
2. Log in with admin credentials
3. Click "New Assessment"
4. Configure target and assessment parameters:
   • Target: IP address or hostname
   • Phases: Select assessment phases (Scanning, Exploitation, etc.)
   • Tools: Choose security tools to use
   • Intensity: Light, Normal, Aggressive
5. Click "Start Assessment"
6. Monitor real-time progress
7. Download report when complete

Via API

# Authenticate
curl -X POST http://localhost:5000/api/v1/auth/login \
  -H "Content-Type: application/json" \
  -d '{
    "username": "admin",
    "password": "your_password"
  }'

# Store JWT token from response
TOKEN="your_jwt_token_here"

# Create assessment
curl -X POST http://localhost:5000/api/v1/assessments \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "target": "192.168.1.100",
    "name": "Test Assessment",
    "phases": ["reconnaissance", "scanning", "enumeration"],
    "tools": ["nmap", "nikto"],
    "intensity": "normal"
  }'

# Get assessment status
curl -X GET http://localhost:5000/api/v1/assessments/{id} \
  -H "Authorization: Bearer $TOKEN"

# Download PDF report
curl -X GET http://localhost:5000/api/v1/assessments/{id}/report/pdf \
  -H "Authorization: Bearer $TOKEN" \
  --output report.pdf

Via CLI

# Quick scan
python cli.py assess --target 192.168.1.100 --quick

# Full assessment with all phases
python cli.py assess --target 192.168.1.100 --full

# Custom phases
python cli.py assess --target 192.168.1.100 \
  --phases scanning,enumeration,exploitation

# List all assessments
python cli.py list-assessments

# Get assessment report
python cli.py get-report --id <assessment_id> --format pdf

🔧 Development

Project Structure Explained

Core Modules

core/orchestrator/enhanced_orchestrator.py

• Main orchestration engine
• Coordinates all assessment phases
• Manages tool execution
• Handles error recovery and retries

core/orchestrator/finding_aggregator.py

• Consolidates findings from all phases
• Removes duplicates and correlates related findings
• Integrates baseline configuration findings
• Generates unified finding reports

core/scanning/enhanced_vulnerability_scanner.py

• Service-level vulnerability detection
• Protocol-specific security checks (SSH, FTP, SMB, HTTP, etc.)
• Weak configuration detection
• CVE mapping

core/vulnerability/baseline_checker.py

• Host-level configuration assessment
• End-of-life OS detection
• Firewall/antivirus status checking
• SMBv1 and dangerous protocol detection
• Zero-dependency vulnerability detection

Adding New Features

Adding a New Baseline Check

Edit core/vulnerability/baseline_checker.py:

def check_weak_passwords(scan_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """
    Check if system has weak password policy.
    """
    password_policy = scan_data.get('password_policy', {})
    min_length = password_policy.get('min_length', 0)
    
    if min_length < 12:
        return {
            'vulnerability_id': 'weak_password_policy',
            'title': 'Weak Password Policy Configured',
            'description': f'System allows passwords shorter than 12 characters (current: {min_length})',
            'severity': 'HIGH',
            'cvss_score': 7.5,
            'cve_ids': [],
            'category': 'authentication',
            'exploitability': 'HIGH',
            'remediation': 'Configure password policy: minimum 12 characters, complexity requirements, expiration',
            'compliance_violations': [
                'PCI_DSS_8.2.3',
                'NIST_800_53_IA_5',
                'ISO_27001_A.9.4.3'
            ]
        }
    return None

# Add to BASELINE_CHECKS list
BASELINE_CHECKS = [
    check_eol_operating_system,
    check_firewall_disabled,
    check_no_antivirus,
    check_no_updates,
    check_smb_v1_enabled,
    check_dangerous_ports,
    check_weak_passwords,  # Your new check
]

Adding a New Exploitation Module

Create core/exploitation/custom_exploit.py:

import logging
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)

class CustomExploit:
    """
    Custom exploitation module.
    """
    
    def __init__(self, target: str):
        self.target = target
        self.exploited = False
        
    def check_vulnerability(self) -> bool:
        """
        Check if target is vulnerable.
        """
        # Implement vulnerability check
        return True
        
    def exploit(self) -> Dict[str, Any]:
        """
        Attempt exploitation.
        """
        if not self.check_vulnerability():
            return {
                'success': False,
                'reason': 'Target not vulnerable'
            }
            
        # Implement exploitation logic
        try:
            # Your exploitation code here
            self.exploited = True
            return {
                'success': True,
                'access_level': 'user',
                'method': 'custom_exploit',
                'evidence': {}
            }
        except Exception as e:
            logger.error(f"Exploitation failed: {e}")
            return {
                'success': False,
                'reason': str(e)
            }

Register in core/orchestrator/enhanced_orchestrator.py:

from core.exploitation.custom_exploit import CustomExploit

# In orchestrator's exploitation phase
exploit_module = CustomExploit(target)
result = exploit_module.exploit()

Adding a New Report Format

Create api/export/custom_export.py:

import logging
from typing import Dict, Any
import io

logger = logging.getLogger(__name__)

def generate_custom_report(assessment_data: Dict[str, Any]) -> bytes:
    """
    Generate custom format report.
    """
    output = io.BytesIO()
    
    # Implement your report generation logic
    report_content = format_custom_report(assessment_data)
    output.write(report_content.encode('utf-8'))
    
    output.seek(0)
    return output.read()

def format_custom_report(data: Dict[str, Any]) -> str:
    """
    Format report content.
    """
    # Your custom formatting logic
    return "Custom Report Content"

Add route in api/export/routes.py:

from api.export.custom_export import generate_custom_report

@bp.route('/assessments/<int:assessment_id>/report/custom', methods=['GET'])
@jwt_required()
def export_custom_report(assessment_id):
    """Export assessment report in custom format."""
    assessment = Assessment.query.get_or_404(assessment_id)
    report_data = assessment.get_report_data()
    
    report_bytes = generate_custom_report(report_data)
    
    return send_file(
        io.BytesIO(report_bytes),
        mimetype='application/custom',
        as_attachment=True,
        download_name=f'assessment_{assessment_id}_report.custom'
    )

Testing

Running Tests

# Run all tests
pytest tests/ -v

# Run with coverage
pytest tests/ -v --cov=core --cov=api --cov-report=html

# Run specific test file
pytest tests/test_baseline_integration.py -v

# Run baseline checker integration test
python tests/test_baseline_integration.py

Writing Tests

Create tests/test_custom_feature.py:

import pytest
from core.custom_feature import CustomFeature

@pytest.fixture
def custom_feature():
    """Fixture for custom feature."""
    return CustomFeature()

def test_custom_feature_basic(custom_feature):
    """Test basic functionality."""
    result = custom_feature.process()
    assert result is not None
    assert result['status'] == 'success'

def test_custom_feature_error_handling(custom_feature):
    """Test error handling."""
    with pytest.raises(ValueError):
        custom_feature.process(invalid_input=True)

Code Quality

# Linting (check code style)
flake8 core/ api/ --max-line-length=120

# Formatting (auto-format code)
black core/ api/ tests/

# Type checking
mypy core/ api/ --strict

# Security scanning
bandit -r core/ api/

Database Migrations

# Create new migration
alembic revision --autogenerate -m "Add new field to assessment table"

# Review generated migration in migrations/versions/
# Edit if necessary

# Apply migration
alembic upgrade head

# Rollback migration
alembic downgrade -1

# View migration history
alembic history

# Check current version
alembic current

📊 API Reference

Authentication

POST   /api/v1/auth/login       # Authenticate user
POST   /api/v1/auth/refresh     # Refresh JWT token
POST   /api/v1/auth/logout      # Logout user
POST   /api/v1/auth/register    # Register new user (if enabled)

Assessments

GET    /api/v1/assessments              # List all assessments
POST   /api/v1/assessments              # Create new assessment
GET    /api/v1/assessments/:id          # Get assessment details
PATCH  /api/v1/assessments/:id          # Update assessment
DELETE /api/v1/assessments/:id          # Delete assessment
POST   /api/v1/assessments/:id/start    # Start assessment
POST   /api/v1/assessments/:id/stop     # Stop assessment
GET    /api/v1/assessments/:id/status   # Get real-time status

Findings

GET    /api/v1/assessments/:id/findings              # List findings
GET    /api/v1/assessments/:id/findings/:finding_id  # Get finding details
PATCH  /api/v1/assessments/:id/findings/:finding_id  # Update finding

Reports

GET /api/v1/assessments/:id/report/pdf   # Export PDF report
GET /api/v1/assessments/:id/report/csv   # Export CSV report
GET /api/v1/assessments/:id/report/json  # Export JSON report

Health & Metrics

GET /api/v1/health   # System health check
GET /api/v1/metrics  # System metrics

Full API documentation: /api/v1/docs (Swagger UI when application is running)

🔒 Security Considerations

Production Deployment Checklist

• Use HTTPS (TLS 1.2+ only)
• Generate strong JWT secret keys (32+ characters, cryptographically random)
• Use strong database passwords (16+ characters)
• Configure firewall rules (restrict access to trusted networks)
• Enable audit logging
• Implement regular backup strategy
• Keep dependencies updated (pip list --outdated)
• Disable debug mode (FLASK_DEBUG=False)
• Configure rate limiting
• Review and restrict CORS origins
• Implement IP whitelisting for API access
• Enable PostgreSQL SSL connections
• Use environment variables for secrets (never commit to git)
• Regular security updates for OS and tools (Nmap, Nikto, etc.)

Legal and Ethical Use

This framework must only be used for:

• Authorized security assessments with written permission
• Educational purposes in isolated lab environments
• Research with explicit authorization
• Compliance auditing with proper authorization

Illegal Use Includes:

• Testing systems without explicit written authorization
• Accessing data you're not authorized to access
• Causing damage or disruption to systems
• Any activity that violates computer crime laws in your jurisdiction

🐛 Troubleshooting

Application Won't Start

# Check logs
Get-Content logs/application.log -Tail 50 -Wait

# Verify database connection
python cli.py test-db

# Check Redis connection
redis-cli ping

# Verify Python version
python --version  # Should be 3.11+

Database Connection Issues

# Test PostgreSQL connection
Test-NetConnection -ComputerName localhost -Port 5432

# Check PostgreSQL service (Windows)
Get-Service postgresql*

# Check PostgreSQL service (Linux)
sudo systemctl status postgresql

# Verify database exists
psql -U postgres -c "\l"

Assessment Fails to Start

# Check Celery worker status
celery -A main.celery inspect active

# View Celery logs
Get-Content logs/celery.log -Tail 50 -Wait

# Verify tool availability
nmap --version
nikto -Version
sqlmap --version

No Baseline Findings Generated

# Enable debug logging in Python shell
import logging
logging.basicConfig(level=logging.DEBUG)

# Test baseline checker directly
from core.vulnerability.baseline_checker import generate_baseline_findings

test_data = {
    'os_name': 'Microsoft Windows XP',
    'firewall_enabled': False,
    'antivirus_present': False,
    'smb_v1_enabled': True
}

findings = generate_baseline_findings(test_data)
print(f"Generated {len(findings)} findings")
for f in findings:
    print(f["  - {f['title']} (CVSS: {f['cvss_score']})"]

Performance Issues

• Increase Celery worker concurrency: celery -A main.celery worker --concurrency=8
• Enable Redis caching in .env
• Tune PostgreSQL configuration (postgresql.conf)
• Check system resources: CPU, RAM, disk I/O
• Review assessment intensity settings (reduce from "aggressive" to "normal")

📚 Documentation

• Baseline Vulnerability Checker Guide - Comprehensive guide to baseline configuration assessment
• API Documentation - Full API reference with examples
• Development Guide - Detailed development workflows
• Windows Deployment - Windows-specific deployment instructions
• Linux Deployment - Linux deployment guide

🤝 Contributing

Contributions are welcome! Please follow these steps:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes with clear commit messages
4. Add tests for new functionality
5. Ensure all tests pass (pytest tests/)
6. Run code quality checks (flake8, black, mypy)
7. Update documentation as needed
8. Submit a Pull Request

Contribution Guidelines

• Follow PEP 8 style guidelines
• Write clear, descriptive commit messages
• Add docstrings to all functions and classes
• Include type hints for function parameters and return values
• Write tests for new features
• Update README.md if adding user-facing features
• Ensure backward compatibility

📜 License

MIT License - see LICENSE file for details.

🙏 Acknowledgments

• Built with Flask, SQLAlchemy, Celery, and React
• Integrates Nmap, Nikto, SQLMap, and other industry-standard tools
• CEH v12 methodology alignment
• NIST Cybersecurity Framework mapping
• OWASP Top 10 integration

📞 Support

For issues and questions:

• GitHub Issues: https://github.com/zy538324/TM470-F1746464/issues
• Documentation: See /docs directory for detailed guides
• Security Issues: Report via private vulnerability disclosure (GitHub Security tab)

📈 Project Status

Status: Educational Project - Production-Ready Code
Version: 1.0
Last Updated: March 2026
Maintainer: Matt Palmer

---

⚠️ Legal Notice: This tool is intended for authorized security testing only. The developers assume no liability for misuse. Always obtain explicit written permission before testing any systems you do not own.