Port Analysis

Analyze and interpret port scan results for security assessment.

Installation

pip install pandas tabulate

Well-Known Ports Reference

WELL_KNOWN_PORTS = {
    # Remote Access
    22: {'service': 'SSH', 'category': 'remote-access', 'risk': 'medium'},
    23: {'service': 'Telnet', 'category': 'remote-access', 'risk': 'high'},
    3389: {'service': 'RDP', 'category': 'remote-access', 'risk': 'high'},
    5900: {'service': 'VNC', 'category': 'remote-access', 'risk': 'high'},

    # Web Services
    80: {'service': 'HTTP', 'category': 'web', 'risk': 'medium'},
    443: {'service': 'HTTPS', 'category': 'web', 'risk': 'low'},
    8080: {'service': 'HTTP-Alt', 'category': 'web', 'risk': 'medium'},
    8443: {'service': 'HTTPS-Alt', 'category': 'web', 'risk': 'low'},

    # Database
    3306: {'service': 'MySQL', 'category': 'database', 'risk': 'critical'},
    5432: {'service': 'PostgreSQL', 'category': 'database', 'risk': 'critical'},
    1433: {'service': 'MSSQL', 'category': 'database', 'risk': 'critical'},
    1521: {'service': 'Oracle', 'category': 'database', 'risk': 'critical'},
    27017: {'service': 'MongoDB', 'category': 'database', 'risk': 'critical'},
    6379: {'service': 'Redis', 'category': 'database', 'risk': 'critical'},

    # File Sharing
    21: {'service': 'FTP', 'category': 'file-sharing', 'risk': 'high'},
    445: {'service': 'SMB', 'category': 'file-sharing', 'risk': 'critical'},
    139: {'service': 'NetBIOS', 'category': 'file-sharing', 'risk': 'high'},
    2049: {'service': 'NFS', 'category': 'file-sharing', 'risk': 'high'},

    # Mail
    25: {'service': 'SMTP', 'category': 'mail', 'risk': 'medium'},
    110: {'service': 'POP3', 'category': 'mail', 'risk': 'medium'},
    143: {'service': 'IMAP', 'category': 'mail', 'risk': 'medium'},
    465: {'service': 'SMTPS', 'category': 'mail', 'risk': 'low'},
    993: {'service': 'IMAPS', 'category': 'mail', 'risk': 'low'},

    # Directory Services
    389: {'service': 'LDAP', 'category': 'directory', 'risk': 'high'},
    636: {'service': 'LDAPS', 'category': 'directory', 'risk': 'medium'},
    88: {'service': 'Kerberos', 'category': 'directory', 'risk': 'medium'},

    # Other
    53: {'service': 'DNS', 'category': 'infrastructure', 'risk': 'medium'},
    161: {'service': 'SNMP', 'category': 'management', 'risk': 'high'},
    111: {'service': 'RPCBind', 'category': 'infrastructure', 'risk': 'medium'},
}

def get_port_info(port: int) -> dict:
    """Get information about a port."""
    return WELL_KNOWN_PORTS.get(port, {
        'service': 'unknown',
        'category': 'unknown',
        'risk': 'unknown'
    })

Parse Scan Results

import xml.etree.ElementTree as ET
import json

def parse_nmap_xml(xml_file: str) -> list:
    """Parse Nmap XML output."""
    tree = ET.parse(xml_file)
    root = tree.getroot()

    results = []
    for host in root.findall('host'):
        host_data = {
            'ip': None,
            'hostname': None,
            'state': None,
            'ports': []
        }

        # Get IP address
        for addr in host.findall('address'):
            if addr.get('addrtype') == 'ipv4':
                host_data['ip'] = addr.get('addr')

        # Get hostname
        hostnames = host.find('hostnames')
        if hostnames is not None:
            hostname = hostnames.find('hostname')
            if hostname is not None:
                host_data['hostname'] = hostname.get('name')

        # Get state
        status = host.find('status')
        if status is not None:
            host_data['state'] = status.get('state')

        # Get ports
        ports = host.find('ports')
        if ports is not None:
            for port in ports.findall('port'):
                port_data = {
                    'port': int(port.get('portid')),
                    'protocol': port.get('protocol'),
                    'state': port.find('state').get('state'),
                    'service': port.find('service').get('name') if port.find('service') is not None else 'unknown',
                    'product': port.find('service').get('product', '') if port.find('service') is not None else '',
                    'version': port.find('service').get('version', '') if port.find('service') is not None else '',
                }
                host_data['ports'].append(port_data)

        results.append(host_data)

    return results

def parse_gnmap(gnmap_file: str) -> list:
    """Parse Nmap grepable output."""
    results = []

    with open(gnmap_file, 'r') as f:
        for line in f:
            if 'Ports:' in line:
                parts = line.split('\t')
                host = parts[0].split()[1]

                ports_section = [p for p in parts if p.startswith('Ports:')][0]
                ports_str = ports_section.replace('Ports: ', '')

                ports = []
                for port_info in ports_str.split(', '):
                    fields = port_info.split('/')
                    if len(fields) >= 5:
                        ports.append({
                            'port': int(fields[0]),
                            'state': fields[1],
                            'protocol': fields[2],
                            'service': fields[4]
                        })

                results.append({'host': host, 'ports': ports})

    return results

Categorize and Prioritize

import pandas as pd

def categorize_ports(scan_results: list) -> pd.DataFrame:
    """Categorize ports by service type and risk."""
    rows = []

    for host in scan_results:
        for port in host.get('ports', []):
            if port['state'] == 'open':
                info = get_port_info(port['port'])
                rows.append({
                    'host': host.get('ip', host.get('host')),
                    'port': port['port'],
                    'protocol': port.get('protocol', 'tcp'),
                    'service': port.get('service', info['service']),
                    'product': port.get('product', ''),
                    'version': port.get('version', ''),
                    'category': info['category'],
                    'risk': info['risk']
                })

    df = pd.DataFrame(rows)
    return df

def prioritize_targets(df: pd.DataFrame) -> pd.DataFrame:
    """Prioritize targets by risk level."""
    risk_order = {'critical': 0, 'high': 1, 'medium': 2, 'low': 3, 'unknown': 4}
    df['risk_score'] = df['risk'].map(risk_order)
    return df.sort_values('risk_score')

def get_attack_surface(df: pd.DataFrame) -> dict:
    """Summarize attack surface by category."""
    summary = {
        'total_hosts': df['host'].nunique(),
        'total_open_ports': len(df),
        'by_category': df.groupby('category')['port'].count().to_dict(),
        'by_risk': df.groupby('risk')['port'].count().to_dict(),
        'critical_services': df[df['risk'] == 'critical'][['host', 'port', 'service']].to_dict('records')
    }
    return summary

Generate Reports

from tabulate import tabulate

def generate_port_report(df: pd.DataFrame) -> str:
    """Generate human-readable port report."""
    report = []

    report.append("=" * 60)
    report.append("PORT ANALYSIS REPORT")
    report.append("=" * 60)

    # Summary
    report.append(f"\nTotal Hosts: {df['host'].nunique()}")
    report.append(f"Total Open Ports: {len(df)}")

    # Risk summary
    report.append("\nRisk Summary:")
    for risk in ['critical', 'high', 'medium', 'low']:
        count = len(df[df['risk'] == risk])
        report.append(f"  {risk.upper()}: {count}")

    # Critical findings
    critical = df[df['risk'] == 'critical']
    if not critical.empty:
        report.append("\n" + "!" * 60)
        report.append("CRITICAL FINDINGS")
        report.append("!" * 60)
        report.append(tabulate(
            critical[['host', 'port', 'service', 'product']],
            headers='keys',
            tablefmt='grid'
        ))

    # By category
    report.append("\n" + "-" * 60)
    report.append("SERVICES BY CATEGORY")
    report.append("-" * 60)
    for category in df['category'].unique():
        cat_df = df[df['category'] == category]
        report.append(f"\n{category.upper()}:")
        report.append(tabulate(
            cat_df[['host', 'port', 'service', 'version']],
            headers='keys',
            tablefmt='simple'
        ))

    return '\n'.join(report)

def export_to_csv(df: pd.DataFrame, filename: str):
    """Export analysis to CSV."""
    df.to_csv(filename, index=False)

def export_to_json(df: pd.DataFrame, filename: str):
    """Export analysis to JSON."""
    df.to_json(filename, orient='records', indent=2)

Identify Potential Vulnerabilities

VERSION_VULNERABILITIES = {
    'OpenSSH': {
        '7.2': ['CVE-2016-6515', 'CVE-2016-6210'],
        '6.9': ['CVE-2015-5600'],
    },
    'Apache': {
        '2.4.49': ['CVE-2021-41773'],
        '2.4.50': ['CVE-2021-42013'],
    },
    'vsftpd': {
        '2.3.4': ['CVE-2011-2523'],  # Backdoor
    }
}

def check_known_vulnerabilities(df: pd.DataFrame) -> list:
    """Check for known vulnerable versions."""
    vulnerabilities = []

    for _, row in df.iterrows():
        product = row.get('product', '')
        version = row.get('version', '')

        for vuln_product, versions in VERSION_VULNERABILITIES.items():
            if vuln_product.lower() in product.lower():
                for vuln_version, cves in versions.items():
                    if vuln_version in version:
                        vulnerabilities.append({
                            'host': row['host'],
                            'port': row['port'],
                            'product': product,
                            'version': version,
                            'cves': cves
                        })

    return vulnerabilities

def identify_misconfigurations(df: pd.DataFrame) -> list:
    """Identify potential misconfigurations."""
    issues = []

    # Check for unencrypted services
    unencrypted = df[df['service'].isin(['telnet', 'ftp', 'http', 'ldap'])]
    for _, row in unencrypted.iterrows():
        issues.append({
            'host': row['host'],
            'port': row['port'],
            'issue': f"Unencrypted {row['service']} service exposed",
            'severity': 'medium'
        })

    # Check for exposed databases
    databases = df[df['category'] == 'database']
    for _, row in databases.iterrows():
        issues.append({
            'host': row['host'],
            'port': row['port'],
            'issue': f"Database {row['service']} exposed to network",
            'severity': 'critical'
        })

    return issues