Why Annual Pentesting Fails Fast-Moving Teams (And What Replaces It)

import datetime
from dataclasses import dataclass, field
from typing import List, Dict
import statistics

@dataclass
class DeploymentEvent:
    date: datetime.date
    commit_count: int
    files_changed: int
    security_relevant_changes: int  # Auth, API, DB layer changes
    introduced_vulnerabilities: int  # Known from retrospective analysis

@dataclass
class SecurityTestEvent:
    date: datetime.date
    test_type: str  # 'annual', 'quarterly', 'monthly', 'continuous'
    findings_count: int
    critical_findings: int

def calculate_deployment_velocity_gap(
    deployment_history: List[DeploymentEvent],
    security_test_history: List[SecurityTestEvent],
    year: int
) -> Dict:
    """
    Calculate the actual deployment velocity gap for an organization.
    Uses historical deployment and testing data to quantify exposure windows.
    """

    # Filter to the analysis year
    deployments = [d for d in deployment_history
                   if d.date.year == year]
    tests = [t for t in security_test_history
             if t.date.year == year]
    tests.sort(key=lambda x: x.date)

    if not tests:
        return {'error': 'No security tests in period'}

    # Calculate testing intervals
    test_dates = [t.date for t in tests]
    intervals = []

    if len(test_dates) > 1:
        for i in range(1, len(test_dates)):
            gap_days = (test_dates[i] - test_dates[i-1]).days
            intervals.append(gap_days)

    # Add gap from last test to end of year (remaining exposure)
    year_end = datetime.date(year, 12, 31)
    remaining = (year_end - test_dates[-1]).days
    intervals.append(remaining)

    # Calculate deployments in each gap window
    gap_analysis = []

    window_start = datetime.date(year, 1, 1)
    for i, test in enumerate(tests):
        window_end = test.date

        # Count deployments in this window
        window_deployments = [
            d for d in deployments
            if window_start <= d.date < window_end
        ]

        security_changes = sum(d.security_relevant_changes for d in window_deployments)
        total_commits = sum(d.commit_count for d in window_deployments)

        # Estimate vulnerability introduction rate
        # Industry baseline: ~1 security-relevant bug per 1000 LOC changed
        # Security-relevant changes have higher density
        estimated_vulns_introduced = security_changes * 0.05  # 5% introduce a finding

        gap_analysis.append({
            'window_start': window_start.isoformat(),
            'window_end': window_end.isoformat(),
            'gap_days': (window_end - window_start).days,
            'deployments_in_window': len(window_deployments),
            'total_commits': total_commits,
            'security_relevant_changes': security_changes,
            'estimated_vulns_introduced': round(estimated_vulns_introduced, 1),
            'test_at_end': test.test_type,
            'findings_at_test': test.findings_count
        })

        window_start = test.date

    # Calculate aggregate metrics
    total_gap_days = sum(g['gap_days'] for g in gap_analysis)
    max_gap = max(g['gap_days'] for g in gap_analysis)
    avg_gap = statistics.mean(g['gap_days'] for g in gap_analysis)
    total_security_changes_untested = sum(
        g['security_relevant_changes'] for g in gap_analysis
    )

    return {
        'year': year,
        'total_deployments': len(deployments),
        'total_security_tests': len(tests),
        'testing_intervals': intervals,
        'max_exposure_window_days': max_gap,
        'average_exposure_window_days': round(avg_gap, 0),
        'total_security_relevant_changes': sum(
            d.security_relevant_changes for d in deployments
        ),
        'estimated_total_vulns_introduced': round(
            sum(d.security_relevant_changes for d in deployments) * 0.05, 0
        ),
        'gap_analysis': gap_analysis,
        'risk_assessment': classify_gap_risk(max_gap, avg_gap)
    }

def classify_gap_risk(max_gap: int, avg_gap: float) -> str:
    if max_gap > 180 or avg_gap > 90:
        return 'CRITICAL — Unacceptably long exposure windows'
    elif max_gap > 90 or avg_gap > 45:
        return 'HIGH — Significant exposure windows between tests'
    elif max_gap > 30 or avg_gap > 14:
        return 'MEDIUM — Manageable with process discipline'
    else:
        return 'LOW — Continuous testing adequately covers deployment velocity'

@dataclass
class AttackSurfaceSnapshot:
    """Represents the attack surface at a point in time"""
    date: datetime.date
    api_endpoints: set          # All discoverable API endpoints
    auth_mechanisms: set        # Auth patterns in use (JWT, session, API key)
    external_integrations: set  # Third-party services integrated
    open_ports: set             # Exposed network ports
    subdomains: set             # Discoverable subdomains
    javascript_bundles: list    # Frontend bundle hashes
    dependencies: dict          # {package: version} dependency map
    service_accounts: set       # Cloud/K8s service accounts

def calculate_attack_surface_drift(
    snapshot_a: AttackSurfaceSnapshot,
    snapshot_b: AttackSurfaceSnapshot
) -> Dict:
    """
    Quantify how much the attack surface changed between two snapshots.
    Each change is a potential security regression point.
    """

    drift = {}

    # New endpoints added since last test
    new_endpoints = snapshot_b.api_endpoints - snapshot_a.api_endpoints
    removed_endpoints = snapshot_a.api_endpoints - snapshot_b.api_endpoints

    drift['api_endpoints'] = {
        'added': list(new_endpoints),
        'removed': list(removed_endpoints),
        'net_change': len(new_endpoints) - len(removed_endpoints),
        'security_risk': 'Each new endpoint is untested attack surface',
        'high_risk_additions': [
            ep for ep in new_endpoints
            if any(word in ep.lower() for word in
                   ['admin', 'export', 'delete', 'upload', 'import', 'webhook'])
        ]
    }

    # New authentication mechanisms
    new_auth = snapshot_b.auth_mechanisms - snapshot_a.auth_mechanisms
    drift['auth_mechanisms'] = {
        'added': list(new_auth),
        'security_risk': 'New auth mechanisms may have configuration vulnerabilities'
            if new_auth else 'None',
    }

    # New external integrations (new credential chains)
    new_integrations = snapshot_b.external_integrations - snapshot_a.external_integrations
    drift['external_integrations'] = {
        'added': list(new_integrations),
        'security_risk': 'Each integration adds credential exposure surface and SSRF targets',
        'count': len(new_integrations)
    }

    # New subdomains (potential subdomain takeover targets, broader CORS surface)
    new_subdomains = snapshot_b.subdomains - snapshot_a.subdomains
    drift['subdomains'] = {
        'added': list(new_subdomains),
        'count': len(new_subdomains)
    }

    # Dependency changes (new CVE exposure)
    dep_changes = {}
    for pkg, version in snapshot_b.dependencies.items():
        if pkg not in snapshot_a.dependencies:
            dep_changes[pkg] = {'status': 'added', 'version': version}
        elif snapshot_a.dependencies[pkg] != version:
            dep_changes[pkg] = {
                'status': 'updated',
                'old_version': snapshot_a.dependencies[pkg],
                'new_version': version
            }

    for pkg in snapshot_a.dependencies:
        if pkg not in snapshot_b.dependencies:
            dep_changes[pkg] = {'status': 'removed'}

    drift['dependencies'] = {
        'changes': dep_changes,
        'new_packages': sum(1 for v in dep_changes.values() if v['status'] == 'added'),
        'updated_packages': sum(1 for v in dep_changes.values() if v['status'] == 'updated'),
        'removed_packages': sum(1 for v in dep_changes.values() if v['status'] == 'removed')
    }

    # Risk score for the drift period
    drift_risk_score = (
        len(new_endpoints) * 3 +
        len(drift['api_endpoints']['high_risk_additions']) * 5 +
        len(new_auth) * 8 +
        len(new_integrations) * 4 +
        len(new_subdomains) * 2 +
        drift['dependencies']['new_packages'] * 1
    )

    drift['overall_drift_risk_score'] = drift_risk_score
    drift['risk_classification'] = (
        'CRITICAL' if drift_risk_score > 50 else
        'HIGH' if drift_risk_score > 25 else
        'MEDIUM' if drift_risk_score > 10 else
        'LOW'
    )
    drift['days_since_last_test'] = (snapshot_b.date - snapshot_a.date).days

    return drift

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'])

import pytest
import requests
import concurrent.futures
from typing import List, Tuple

class SecurityRegressionSuite:
    """
    Automated security regression tests that run on every deployment.
    Tests for the most common high-impact vulnerabilities.
    These don't replace manual testing — they catch regressions fast.
    """

    def __init__(self, base_url: str, auth_tokens: dict):
        self.base_url = base_url
        self.auth_tokens = auth_tokens  # {user_type: token}
        self.session = requests.Session()

    # ═══════════════════════════════════════════════════════
    # AUTHENTICATION REGRESSION TESTS
    # ═══════════════════════════════════════════════════════

    def test_endpoint_requires_authentication(self, endpoints: List[str]) -> List[dict]:
        """Verify all non-public endpoints return 401 without auth"""
        failures = []

        for endpoint in endpoints:
            response = self.session.get(
                f"{self.base_url}{endpoint}",
                allow_redirects=False
            )

            if response.status_code not in [401, 403, 302]:
                failures.append({
                    'test': 'authentication_required',
                    'endpoint': endpoint,
                    'expected': '401/403',
                    'actual': response.status_code,
                    'severity': 'CRITICAL',
                    'finding': f'{endpoint} accessible without authentication'
                })

        return failures

    def test_jwt_algorithm_enforcement(self) -> dict:
        """Test that JWT algorithm is strictly enforced"""
        import jwt
        import json
        import base64

        # Forge a none-algorithm token
        header = base64.urlsafe_b64encode(
            json.dumps({"alg": "none", "typ": "JWT"}).encode()
        ).rstrip(b'=').decode()

        payload = base64.urlsafe_b64encode(
            json.dumps({
                "user_id": "1",
                "role": "admin",
                "exp": 9999999999
            }).encode()
        ).rstrip(b'=').decode()

        forged_token = f"{header}.{payload}."

        response = self.session.get(
            f"{self.base_url}/api/v1/users/profile",
            headers={"Authorization": f"Bearer {forged_token}"}
        )

        return {
            'test': 'jwt_algorithm_none',
            'passed': response.status_code in [401, 403],
            'status_code': response.status_code,
            'severity': 'CRITICAL' if response.status_code == 200 else None,
            'finding': 'JWT none algorithm accepted' if response.status_code == 200 else None
        }

    # ═══════════════════════════════════════════════════════
    # AUTHORIZATION REGRESSION TESTS
    # ═══════════════════════════════════════════════════════

    def test_cross_user_idor(
        self,
        resource_endpoints: List[str],
        user_a_token: str,
        user_b_resources: List[str]
    ) -> List[dict]:
        """Test that User A cannot access User B's resources"""
        failures = []

        headers_a = {"Authorization": f"Bearer {user_a_token}"}

        for endpoint_pattern in resource_endpoints:
            for resource_id in user_b_resources:
                url = endpoint_pattern.replace('{id}', resource_id)

                response = self.session.get(
                    f"{self.base_url}{url}",
                    headers=headers_a,
                    allow_redirects=False
                )

                if response.status_code == 200:
                    failures.append({
                        'test': 'cross_user_idor',
                        'endpoint': url,
                        'accessing_user': 'user_a',
                        'resource_owner': 'user_b',
                        'status_code': 200,
                        'severity': 'HIGH',
                        'finding': f'IDOR: User A can access User B resource at {url}'
                    })

        return failures

    def test_horizontal_privilege_escalation(
        self,
        admin_endpoints: List[str],
        standard_user_token: str
    ) -> List[dict]:
        """Test that standard users cannot access admin endpoints"""
        failures = []

        headers = {"Authorization": f"Bearer {standard_user_token}"}

        for endpoint in admin_endpoints:
            response = self.session.get(
                f"{self.base_url}{endpoint}",
                headers=headers,
                allow_redirects=False
            )

            if response.status_code == 200:
                failures.append({
                    'test': 'privilege_escalation',
                    'endpoint': endpoint,
                    'token_type': 'standard_user',
                    'severity': 'CRITICAL',
                    'finding': f'Standard user accesses admin endpoint: {endpoint}'
                })

        return failures

    # ═══════════════════════════════════════════════════════
    # CORS REGRESSION TESTS
    # ═══════════════════════════════════════════════════════

    def test_cors_origin_reflection(self, api_endpoints: List[str]) -> List[dict]:
        """Test that CORS does not reflect arbitrary origins"""
        failures = []

        test_origin = "<https://evil-test-origin-regression.com>"

        for endpoint in api_endpoints:
            response = self.session.get(
                f"{self.base_url}{endpoint}",
                headers={
                    "Origin": test_origin,
                    "Authorization": f"Bearer {self.auth_tokens.get('standard')}"
                },
                allow_redirects=False
            )

            acao = response.headers.get('Access-Control-Allow-Origin', '')
            acac = response.headers.get('Access-Control-Allow-Credentials', '')

            if acao == test_origin and acac.lower() == 'true':
                failures.append({
                    'test': 'cors_origin_reflection',
                    'endpoint': endpoint,
                    'reflected_origin': test_origin,
                    'credentials_allowed': True,
                    'severity': 'CRITICAL',
                    'finding': f'CORS reflects arbitrary origin with credentials at {endpoint}'
                })

        return failures

    # ═══════════════════════════════════════════════════════
    # INJECTION REGRESSION TESTS
    # ═══════════════════════════════════════════════════════

    def test_sql_injection_time_based(
        self,
        injectable_endpoints: List[Tuple[str, str, str]]  # (endpoint, method, param)
    ) -> List[dict]:
        """
        Test for SQL injection using time-based detection.
        Safe — doesn't read any data, only measures response time difference.
        """
        failures = []

        safe_payload = "test_value_regression_123"
        # MySQL time-based: SLEEP(5)
        sqli_payload_mysql = "test' AND SLEEP(3)--"
        # PostgreSQL time-based: pg_sleep
        sqli_payload_pg = "test' AND pg_sleep(3)--"

        for endpoint, method, param in injectable_endpoints:
            # Baseline response time
            start = time.time()
            safe_response = self.session.request(
                method,
                f"{self.base_url}{endpoint}",
                params={param: safe_payload} if method == 'GET' else None,
                json={param: safe_payload} if method == 'POST' else None,
                headers={"Authorization": f"Bearer {self.auth_tokens.get('standard')}"},
                timeout=10
            )
            baseline_time = time.time() - start

            # Test with MySQL payload
            start = time.time()
            sqli_response = self.session.request(
                method,
                f"{self.base_url}{endpoint}",
                params={param: sqli_payload_mysql} if method == 'GET' else None,
                json={param: sqli_payload_mysql} if method == 'POST' else None,
                headers={"Authorization": f"Bearer {self.auth_tokens.get('standard')}"},
                timeout=15
            )
            test_time = time.time() - start

            # If response is significantly delayed → time-based SQL injection
            if test_time > baseline_time + 2.5:  # 2.5s threshold for 3s sleep
                failures.append({
                    'test': 'sql_injection_time_based',
                    'endpoint': endpoint,
                    'parameter': param,
                    'baseline_time': round(baseline_time, 2),
                    'injection_time': round(test_time, 2),
                    'delay': round(test_time - baseline_time, 2),
                    'severity': 'HIGH',
                    'finding': f'Time-based SQL injection at {endpoint} parameter {param}'
                })

        return failures

    def run_full_regression_suite(
        self,
        endpoints: dict,  # {endpoint_type: [endpoints]}
        user_accounts: dict  # {user_type: {token, resources}}
    ) -> dict:
        """Run the complete regression suite and return consolidated results"""

        all_failures = []

        # Authentication tests
        auth_failures = self.test_endpoint_requires_authentication(
            endpoints.get('authenticated', [])
        )
        all_failures.extend(auth_failures)

        jwt_result = self.test_jwt_algorithm_enforcement()
        if not jwt_result['passed']:
            all_failures.append(jwt_result)

        # Authorization tests
        if 'user_a' in user_accounts and 'user_b' in user_accounts:
            idor_failures = self.test_cross_user_idor(
                endpoints.get('resource', []),
                user_accounts['user_a']['token'],
                user_accounts['user_b']['resource_ids']
            )
            all_failures.extend(idor_failures)

        priv_esc_failures = self.test_horizontal_privilege_escalation(
            endpoints.get('admin', []),
            user_accounts.get('standard', {}).get('token', '')
        )
        all_failures.extend(priv_esc_failures)

        # CORS tests
        cors_failures = self.test_cors_origin_reflection(
            endpoints.get('api', [])
        )
        all_failures.extend(cors_failures)

        # Summary
        critical = [f for f in all_failures if f.get('severity') == 'CRITICAL']
        high = [f for f in all_failures if f.get('severity') == 'HIGH']

        return {
            'total_failures': len(all_failures),
            'critical': len(critical),
            'high': len(high),
            'passed': len(all_failures) == 0,
            'failures': all_failures,
            'recommendation': 'BLOCK DEPLOYMENT' if critical else
                             'REVIEW BEFORE DEPLOYMENT' if high else
                             'DEPLOY WITH MONITORING'
        }