Continuous Penetration Testing vs Annual Pentesting: Which Model Is Right for Your SaaS Company?

class SprintSecurityTestingProgram:
    """
    Operationalizes sprint-cadence security testing.
    Each sprint's changed components are tested before the next sprint begins.
    """

    def __init__(self, repo_url: str, pentest_team_contact: str):
        self.repo_url = repo_url
        self.pentest_team = pentest_team_contact
        self.sprint_history = []

    def analyze_sprint_changes(
        self,
        sprint_start: datetime.date,
        sprint_end: datetime.date,
        merged_prs: list
    ) -> dict:
        """
        Analyze what changed in a sprint to determine security testing scope.
        """

        changed_components = {
            'authentication': [],    # Changes to auth logic
            'authorization': [],     # Changes to access control
            'api_endpoints': [],     # New or modified endpoints
            'data_access': [],       # ORM, database query changes
            'external_integrations': [],  # Third-party API changes
            'infrastructure': [],    # IaC, Kubernetes, CI/CD changes
            'dependencies': [],      # package.json, requirements.txt changes
            'configuration': [],     # Config files, environment changes
        }

        security_relevant_prs = []

        for pr in merged_prs:
            files_changed = pr.get('files_changed', [])

            # Classify changes by security relevance
            classifications = []

            for file in files_changed:
                if any(pattern in file.lower() for pattern in [
                    'auth', 'login', 'jwt', 'token', 'session', 'oauth'
                ]):
                    changed_components['authentication'].append(file)
                    classifications.append('authentication')

                elif any(pattern in file.lower() for pattern in [
                    'permission', 'role', 'acl', 'policy', 'rbac', 'middleware'
                ]):
                    changed_components['authorization'].append(file)
                    classifications.append('authorization')

                elif any(pattern in file.lower() for pattern in [
                    'routes', 'views', 'controllers', 'handlers', 'api'
                ]):
                    changed_components['api_endpoints'].append(file)
                    classifications.append('api_endpoints')

                elif any(pattern in file.lower() for pattern in [
                    'models', 'queries', 'repository', 'dao', 'db', 'orm'
                ]):
                    changed_components['data_access'].append(file)
                    classifications.append('data_access')

                elif file in [
                    'package.json', 'package-lock.json', 'requirements.txt',
                    'Pipfile', 'pom.xml', 'build.gradle', 'go.mod'
                ]:
                    changed_components['dependencies'].append(file)
                    classifications.append('dependencies')

                elif any(pattern in file.lower() for pattern in [
                    'kubernetes', 'k8s', 'helm', 'terraform', 'bicep',
                    '.github/workflows', 'jenkinsfile', 'dockerfile'
                ]):
                    changed_components['infrastructure'].append(file)
                    classifications.append('infrastructure')

            if classifications:
                security_relevant_prs.append({
                    'pr_number': pr.get('number'),
                    'title': pr.get('title'),
                    'author': pr.get('author'),
                    'security_categories': list(set(classifications)),
                    'files_changed': len(files_changed),
                    'security_relevant_files': [
                        f for f in files_changed
                        if any(cat in f.lower() for cat in [
                            'auth', 'api', 'model', 'route', 'middleware'
                        ])
                    ]
                })

        # Determine test depth required for this sprint
        risk_score = (
            len(changed_components['authentication']) * 10 +  # Highest weight
            len(changed_components['authorization']) * 8 +
            len(changed_components['api_endpoints']) * 5 +
            len(changed_components['data_access']) * 6 +
            len(changed_components['infrastructure']) * 7 +
            len(changed_components['external_integrations']) * 5 +
            len(changed_components['dependencies']) * 3
        )

        return {
            'sprint_start': sprint_start.isoformat(),
            'sprint_end': sprint_end.isoformat(),
            'total_prs': len(merged_prs),
            'security_relevant_prs': len(security_relevant_prs),
            'changed_components': changed_components,
            'sprint_risk_score': risk_score,
            'recommended_test_depth': self.classify_test_depth(risk_score),
            'estimated_test_hours': self.estimate_test_hours(risk_score),
            'priority_areas': self.identify_priority_areas(changed_components),
            'security_relevant_pr_details': security_relevant_prs
        }

    def classify_test_depth(self, risk_score: int) -> str:
        if risk_score > 100:
            return 'FULL_DEPTH — Authentication changes require complete auth chain review'
        elif risk_score > 50:
            return 'TARGETED_DEEP — Multiple security-relevant changes require deep testing'
        elif risk_score > 20:
            return 'TARGETED_STANDARD — Specific changed components need focused testing'
        else:
            return 'LIGHTWEIGHT — Minor changes, automated testing sufficient'

    def estimate_test_hours(self, risk_score: int) -> str:
        if risk_score > 100:
            return '8–16 hours'
        elif risk_score > 50:
            return '4–8 hours'
        elif risk_score > 20:
            return '2–4 hours'
        else:
            return '1–2 hours'

    def identify_priority_areas(self, changed_components: dict) -> list:
        priorities = []

        if changed_components['authentication']:
            priorities.append({
                'area': 'Authentication',
                'priority': 1,
                'reason': 'Auth changes have highest security impact',
                'test_focus': 'JWT validation, session management, MFA bypass, brute force'
            })

        if changed_components['authorization']:
            priorities.append({
                'area': 'Authorization',
                'priority': 2,
                'reason': 'Access control changes may introduce privilege escalation',
                'test_focus': 'RBAC, IDOR, cross-tenant access, role bypass'
            })

        if changed_components['data_access']:
            priorities.append({
                'area': 'Data Access Layer',
                'priority': 3,
                'reason': 'ORM changes may introduce injection or IDOR',
                'test_focus': 'SQL injection, NoSQL injection, ownership filter presence'
            })

        return sorted(priorities, key=lambda x: x['priority'])

def calculate_tco_comparison(org_profile: dict) -> dict:
    """
    Calculate Total Cost of Ownership for annual vs continuous security testing.
    Includes direct costs, breach probability adjustment, and remediation costs.
    """

    # Organization profile inputs
    annual_revenue = org_profile['annual_revenue']
    deployment_frequency_per_year = org_profile['deployments_per_year']
    engineering_team_size = org_profile['engineering_team_size']
    avg_engineer_hourly_cost = org_profile['avg_engineer_hourly_cost']
    breach_probability_annual = org_profile['estimated_breach_probability']  # e.g., 0.15 = 15%
    avg_breach_cost = org_profile['avg_breach_cost']  # all-in cost if breach occurs

    # ═══════════════════════════════════════════════════════
    # ANNUAL PENETRATION TESTING MODEL
    # ═══════════════════════════════════════════════════════

    annual_model = {}

    # Direct costs
    annual_model['pentest_cost'] = 25000  # Typical annual pentest (1 week, 1-2 testers)
    annual_model['retest_cost'] = 8000    # Retest after remediation

    # Engineering remediation costs
    # Average: 8 findings, 3 days engineering per finding
    avg_findings = 8
    avg_remediation_days = 3
    annual_model['engineering_remediation_cost'] = (
        avg_findings * avg_remediation_days * 8 *  # 8 hours/day
        avg_engineer_hourly_cost
    )

    # Emergency response costs (for critical findings discovered late)
    # Annual model has longer gap → higher probability of undetected critical issue
    # that then requires emergency response
    prob_emergency_response = 0.35  # 35% chance of emergency security incident
    avg_emergency_response_cost = 50000  # War room, hotfix, communication
    annual_model['expected_emergency_response_cost'] = (
        prob_emergency_response * avg_emergency_response_cost
    )

    # Alert fatigue / wasted engineering time on non-exploitable findings
    # Annual test typically has higher percentage of false positives vs continuous
    annual_model['false_positive_remediation_waste'] = (
        avg_findings * 0.3 *  # 30% false positive rate for annual
        2 * 8 *               # 2 days to discover and document it's a false positive
        avg_engineer_hourly_cost
    )

    # Breach risk — adjusted for longer exposure window
    # Annual model has ~180 day average undetected vulnerability window
    # Breach probability scales with exposure window
    exposure_window_days_annual = 180
    annual_model['adjusted_breach_probability'] = breach_probability_annual * (
        exposure_window_days_annual / 365
    )
    annual_model['expected_breach_cost'] = (
        annual_model['adjusted_breach_probability'] * avg_breach_cost
    )

    annual_model['total_direct_cost'] = (
        annual_model['pentest_cost'] +
        annual_model['retest_cost'] +
        annual_model['engineering_remediation_cost'] +
        annual_model['expected_emergency_response_cost'] +
        annual_model['false_positive_remediation_waste']
    )

    annual_model['total_tco'] = (
        annual_model['total_direct_cost'] +
        annual_model['expected_breach_cost']
    )

    # ═══════════════════════════════════════════════════════
    # CONTINUOUS PENETRATION TESTING MODEL
    # ═══════════════════════════════════════════════════════

    continuous_model = {}

    # Direct costs — subscription model
    continuous_model['monthly_subscription'] = 4500  # Typical continuous program
    continuous_model['annual_subscription_cost'] = continuous_model['monthly_subscription'] * 12

    # Engineering remediation costs — findings caught earlier are cheaper to fix
    # Studies show: 6x cheaper to fix in development vs production
    # Continuous testing catches most issues within 2 weeks of introduction
    continuous_avg_findings = 12  # More findings per year (nothing escapes for 11 months)
    continuous_avg_remediation_days = 1.5  # Caught earlier = simpler fix (feature branch)
    continuous_model['engineering_remediation_cost'] = (
        continuous_avg_findings * continuous_avg_remediation_days * 8 *
        avg_engineer_hourly_cost
    )

    # Emergency response costs — much lower (issues caught before breach)
    prob_emergency_response_continuous = 0.08  # 8% vs 35% for annual
    continuous_model['expected_emergency_response_cost'] = (
        prob_emergency_response_continuous * avg_emergency_response_cost
    )

    # Near-zero false positive waste — continuous testing is more targeted
    continuous_model['false_positive_remediation_waste'] = (
        continuous_avg_findings * 0.05 *  # 5% false positive rate
        1 * 8 *
        avg_engineer_hourly_cost
    )

    # Breach risk — dramatically reduced exposure window
    exposure_window_days_continuous = 14  # 2-week sprint cadence
    continuous_model['adjusted_breach_probability'] = breach_probability_annual * (
        exposure_window_days_continuous / 365
    )
    continuous_model['expected_breach_cost'] = (
        continuous_model['adjusted_breach_probability'] * avg_breach_cost
    )

    continuous_model['total_direct_cost'] = (
        continuous_model['annual_subscription_cost'] +
        continuous_model['engineering_remediation_cost'] +
        continuous_model['expected_emergency_response_cost'] +
        continuous_model['false_positive_remediation_waste']
    )

    continuous_model['total_tco'] = (
        continuous_model['total_direct_cost'] +
        continuous_model['expected_breach_cost']
    )

    # Comparison
    tco_savings = annual_model['total_tco'] - continuous_model['total_tco']

    return {
        'organization_profile': org_profile,
        'annual_model': annual_model,
        'continuous_model': continuous_model,
        'comparison': {
            'annual_tco': round(annual_model['total_tco']),
            'continuous_tco': round(continuous_model['total_tco']),
            'tco_savings': round(tco_savings),
            'savings_percentage': round((tco_savings / annual_model['total_tco']) * 100, 1),
            'breakeven_required_breach_probability': (
                annual_model['total_direct_cost'] - continuous_model['total_direct_cost']
            ) / avg_breach_cost,
            'recommendation': 'Continuous' if tco_savings > 0 else 'Annual',
            'primary_savings_driver': (
                'Breach risk reduction' if continuous_model['expected_breach_cost'] <
                annual_model['expected_breach_cost'] * 0.5
                else 'Engineering efficiency'
            )
        }
    }

# Example calculation:
example_org = {
    'annual_revenue': 10_000_000,
    'deployments_per_year': 52,  # Weekly releases
    'engineering_team_size': 15,
    'avg_engineer_hourly_cost': 100,
    'estimated_breach_probability': 0.12,  # 12% annual breach probability
    'avg_breach_cost': 500_000
}

result = calculate_tco_comparison(example_org)
print(f"Annual model TCO:     ${result['comparison']['annual_tco']:,}")
print(f"Continuous model TCO: ${result['comparison']['continuous_tco']:,}")
print(f"Expected savings:     ${result['comparison']['tco_savings']:,}")

class ContinuousSecurityProgramMetrics:
    """Track and report continuous security testing program effectiveness"""

    def calculate_program_kpis(self, program_data: dict) -> dict:

        findings_data = program_data['findings']
        test_events = program_data['test_events']
        deployments = program_data['deployments']

        # KPI 1: Mean Time to Detection (MTTD)
        # How long from vulnerability introduction to detection?
        mttd_values = []
        for finding in findings_data:
            if finding.get('introduction_date') and finding.get('detection_date'):
                days = (finding['detection_date'] - finding['introduction_date']).days
                mttd_values.append(days)

        mttd = sum(mttd_values) / len(mttd_values) if mttd_values else None

        # KPI 2: Mean Time to Remediation (MTTR)
        # How long from detection to confirmed fix?
        mttr_values = []
        for finding in findings_data:
            if finding.get('detection_date') and finding.get('remediation_date'):
                days = (finding['remediation_date'] - finding['detection_date']).days
                mttr_values.append(days)

        mttr = sum(mttr_values) / len(mttr_values) if mttr_values else None

        # KPI 3: Vulnerability Introduction Rate
        # New security findings per 100 deployments
        total_findings = len(findings_data)
        total_deployments = len(deployments)
        vuln_rate = (total_findings / total_deployments * 100) if total_deployments else 0

        # KPI 4: Escape Rate
        # Percentage of vulnerabilities NOT caught before production
        # (Found by external researchers or incident response, not internal testing)
        external_discoveries = sum(
            1 for f in findings_data
            if f.get('discovered_by') == 'external'
        )
        escape_rate = (external_discoveries / total_findings * 100) if total_findings else 0

        # KPI 5: SLA Compliance Rate
        # Percentage of findings remediated within defined SLAs
        sla_compliant = sum(
            1 for f in findings_data
            if f.get('remediated_within_sla') == True
        )
        sla_rate = (sla_compliant / total_findings * 100) if total_findings else 0

        # KPI 6: CVSS Trend
        # Is the average CVSS of findings going up or down over time?
        monthly_avg_cvss = {}
        for finding in findings_data:
            month = finding['detection_date'].strftime('%Y-%m')
            if month not in monthly_avg_cvss:
                monthly_avg_cvss[month] = []
            monthly_avg_cvss[month].append(finding['cvss'])

        cvss_trend = {
            month: sum(scores) / len(scores)
            for month, scores in monthly_avg_cvss.items()
        }

        # KPI 7: Attack Surface Growth Rate
        # How fast is the untested attack surface growing?
        surface_snapshots = program_data.get('surface_snapshots', [])
        if len(surface_snapshots) >= 2:
            first = surface_snapshots[0]
            last = surface_snapshots[-1]
            surface_growth = (
                (len(last['endpoints']) - len(first['endpoints'])) /
                len(first['endpoints']) * 100
            )
        else:
            surface_growth = None

        return {
            'mean_time_to_detection_days': round(mttd, 1) if mttd else 'N/A',
            'mean_time_to_remediation_days': round(mttr, 1) if mttr else 'N/A',
            'vulnerability_introduction_rate_per_100_deployments': round(vuln_rate, 2),
            'escape_rate_percent': round(escape_rate, 1),
            'sla_compliance_rate_percent': round(sla_rate, 1),
            'cvss_trend_by_month': cvss_trend,
            'attack_surface_growth_percent': round(surface_growth, 1) if surface_growth else 'N/A',

            'program_health': self.assess_program_health(mttd, mttr, escape_rate, sla_rate),

            'benchmarks': {
                'mttd_industry_annual': 180,  # days
                'mttd_industry_continuous': 14,
                'mttd_your_program': mttd,
                'mttr_pci_requirement_critical': 1,  # day
                'sla_compliance_target': 95,  # percent
            }
        }

    def assess_program_health(self, mttd, mttr, escape_rate, sla_rate) -> str:
        score = 0

        if mttd and mttd < 14: score += 2
        elif mttd and mttd < 30: score += 1

        if mttr and mttr < 7: score += 2
        elif mttr and mttr < 30: score += 1

        if escape_rate < 5: score += 2
        elif escape_rate < 15: score += 1

        if sla_rate > 95: score += 2
        elif sla_rate > 80: score += 1

        if score >= 7: return 'EXCELLENT'
        elif score >= 5: return 'GOOD'
        elif score >= 3: return 'IMPROVING'
        else: return 'NEEDS_ATTENTION'