How to Pass a Pentest Retest the First Time (Step-by-Step)

# Original finding: IDOR on GET /api/invoices/{id}
# The endpoint returned any invoice by ID without ownership check

# WRONG remediation — fixes only the reported endpoint:
@app.route('/api/invoices/<int:invoice_id>')
@login_required
def get_invoice(invoice_id):
    invoice = Invoice.query.get(invoice_id)
    # Added after finding:
    if invoice.user_id != current_user.id:
        abort(403)
    return jsonify(invoice.to_dict())

# Retest result: ORIGINAL FINDING PASSES — this specific endpoint is fixed
# But variation testing finds:
#   GET /api/invoices/{id}/pdf        ← Same IDOR, different endpoint
#   GET /api/invoices/{id}/line-items ← Same IDOR, sub-resource
#   GET /api/invoices/{id}/history    ← Same IDOR, history endpoint
#   PUT /api/invoices/{id}            ← Same IDOR on update
#   DELETE /api/invoices/{id}         ← Same IDOR on delete

# The fix addressed the symptom (one endpoint) not the cause
# (missing ownership filter pattern across all Invoice operations)

# CORRECT remediation — fixes the root cause:
class InvoiceQuery:
    @staticmethod
    def get_for_current_user(invoice_id):
        """ALWAYS filter by current user — impossible to use without ownership check"""
        return Invoice.query.filter_by(
            id=invoice_id,
            user_id=current_user.id
        ).first_or_404()

# All Invoice endpoints now use InvoiceQuery.get_for_current_user()
# Root cause addressed — not just the reported symptom

# More variation examples that catch incomplete remediations:

# SQL injection fix variation testing:
# Original payload: ' OR '1'='1
# Fixed against: ' OR '1'='1
# Fails against:  ' OR 1=1--
#                 '; WAITFOR DELAY '0:0:5'--
#                 ' UNION SELECT NULL--
#                 admin'--

# JWT bypass fix variation testing:
# Original test: alg:none with empty signature
# Fixed against: alg:none
# Fails against: alg:None (capitalized)
#                alg:NONE (all caps)
#                alg:nOnE (mixed case)

# Path traversal fix variation testing:
# Original payload: ../../../etc/passwd
# Fixed against: ../../../etc/passwd
# Fails against: ..%2F..%2F..%2Fetc%2Fpasswd  (URL encoded)
#                ..%252F..%252Fetc%252Fpasswd  (double encoded)
#                ..%c0%af..%c0%afetc%c0%afpasswd  (overlong UTF-8)

# Original finding: SQL injection via unsanitized user input

# WRONG fix: Input sanitization (security through sanitization)
def get_user(username):
    # Developer sanitizes by removing special characters
    safe_username = re.sub(r"['\\";]", "", username)
    query = f"SELECT * FROM users WHERE username = '{safe_username}'"
    return db.execute(query)

# This "fixes" the SQL injection (partially) but:
# 1. Still vulnerable to some SQL injection variants (comments, UNION, etc.)
# 2. Breaks legitimate usernames with apostrophes ("O'Brien")
# 3. Introduces a second vulnerability: if san_username is logged,
#    the original (un-sanitized) input might be stored/displayed elsewhere

# CORRECT fix: Parameterized queries (actual fix)
def get_user(username):
    query = "SELECT * FROM users WHERE username = %s"
    return db.execute(query, (username,))  # Database handles escaping

# But the wrong fix creates a different attack surface:
# What if the sanitization is applied inconsistently?
# What if there's a code path that bypasses the sanitization?

# Original finding: Race condition on wallet withdrawal
# Fixed with: database-level SELECT FOR UPDATE locking

# But the fix was applied inside a transaction that has
# auto-commit set to True in the ORM configuration:

# WRONG fix — locking doesn't persist across auto-commit transactions
def withdraw_funds(user_id, amount):
    with db.session() as session:
        # SELECT FOR UPDATE acquires lock
        wallet = session.query(Wallet).filter_by(
            user_id=user_id
        ).with_for_update().first()

        if wallet.balance >= amount:
            wallet.balance -= amount
            session.commit()  # Auto-commit releases the lock
            # Another transaction can now also pass the balance check
            # before this commit is visible

# CORRECT fix — lock must be held across the entire check-and-update:
def withdraw_funds(user_id, amount):
    with db.session.begin():  # Explicit transaction boundary
        wallet = db.session.query(Wallet).filter_by(
            user_id=user_id
        ).with_for_update().first()  # Lock acquired

        if wallet.balance >= amount:
            wallet.balance -= amount
            # commit() called implicitly at end of `with` block
            # Lock held until commit
        else:
            raise InsufficientFundsError()

class RetestProtocol:
    """
    Structured retest execution for each finding.
    """

    def retest_finding(self, finding: dict, target_env: str) -> dict:
        """
        Execute retest for a single finding.
        Returns: retest result with pass/fail/partial determination
        """

        result = {
            'finding_id': finding['id'],
            'finding_title': finding['title'],
            'original_cvss': finding['cvss'],
            'retest_date': datetime.now().isoformat(),
            'target_environment': target_env,
            'steps_executed': [],
            'status': None,  # 'REMEDIATED' | 'PARTIALLY_REMEDIATED' | 'NOT_REMEDIATED'
            'new_cvss': None,
            'notes': []
        }

        # Step 1: Reproduce original exploit
        original_exploit_result = self.execute_original_exploit(finding)

        result['steps_executed'].append({
            'step': 'original_exploit_reproduction',
            'description': 'Execute exact attack vector from original finding',
            'payload': finding['exploit']['payload'],
            'expected_outcome': '403/401/400 (vulnerability fixed)',
            'actual_outcome': original_exploit_result['status_code'],
            'passed': original_exploit_result['status_code'] in [401, 403, 400]
        })

        if not result['steps_executed'][-1]['passed']:
            # Original exploit still works — finding NOT remediated
            result['status'] = 'NOT_REMEDIATED'
            result['new_cvss'] = finding['cvss']  # Unchanged
            result['notes'].append('Original exploit vector still active')
            return result

        # Step 2: Variation testing — test related attack vectors
        variation_results = []

        for variation in self.generate_variations(finding):
            var_result = self.execute_variation(variation, finding)
            variation_results.append({
                'variation': variation['name'],
                'payload': variation['payload'],
                'passed': var_result['status_code'] in [401, 403, 400],
                'actual_outcome': var_result['status_code']
            })

        failed_variations = [v for v in variation_results if not v['passed']]
        result['steps_executed'].extend(variation_results)

        # Step 3: Root cause verification
        root_cause_verified = self.verify_root_cause_fix(finding)

        result['steps_executed'].append({
            'step': 'root_cause_verification',
            'description': 'Verify the underlying vulnerability class is addressed',
            'passed': root_cause_verified
        })

        # Determine overall status
        if not failed_variations and root_cause_verified:
            result['status'] = 'REMEDIATED'
            result['new_cvss'] = 0.0
            result['notes'].append('All exploit vectors confirmed fixed')

        elif failed_variations and not root_cause_verified:
            result['status'] = 'NOT_REMEDIATED'
            result['new_cvss'] = finding['cvss']
            result['notes'].append(
                f'Variation testing failed: {[v["variation"] for v in failed_variations]}'
            )

        else:
            # Original fixed but variations fail — partial remediation
            result['status'] = 'PARTIALLY_REMEDIATED'
            # Adjust CVSS based on remaining attack surface
            result['new_cvss'] = self.calculate_adjusted_cvss(
                finding, failed_variations
            )
            result['notes'].append(
                'Original vector fixed but related vulnerabilities remain'
            )

        return result

    def generate_variations(self, finding: dict) -> list:
        """Generate variation test cases based on finding type"""

        finding_type = finding['type']
        variations = []

        if finding_type == 'IDOR':
            # Test adjacent endpoints following same pattern
            base_endpoint = finding['exploit']['endpoint']
            variations = [
                {'name': 'sub_resource',
                 'payload': f"{base_endpoint}/details"},
                {'name': 'list_endpoint',
                 'payload': base_endpoint.replace('{id}', '').rstrip('/')},
                {'name': 'update_method',
                 'method': 'PUT',
                 'payload': base_endpoint},
                {'name': 'delete_method',
                 'method': 'DELETE',
                 'payload': base_endpoint},
            ]

        elif finding_type == 'SQL_INJECTION':
            # Test payload variations
            original_payload = finding['exploit']['payload']
            variations = [
                {'name': 'comment_style', 'payload': "' OR 1=1--"},
                {'name': 'union_based', 'payload': "' UNION SELECT NULL--"},
                {'name': 'time_based', 'payload': "'; WAITFOR DELAY '0:0:5'--"},
                {'name': 'stacked_query', 'payload': "'; DROP TABLE--"},
                {'name': 'double_encoded',
                 'payload': original_payload.replace("'", "%2527")},
            ]

        elif finding_type == 'CORS_MISCONFIGURATION':
            variations = [
                {'name': 'different_attacker_domain',
                 'origin': '<https://different-attacker.net>'},
                {'name': 'null_origin', 'origin': 'null'},
                {'name': 'subdomain_bypass',
                 'origin': f"<https://evil>.{finding['target_domain']}"},
                {'name': 'http_protocol',
                 'origin': finding['allowed_origin'].replace('https://', 'http://')},
            ]

        return variations

def calculate_cvss_delta_report(original_findings: list, retest_results: list) -> dict:
    """
    Calculate the CVSS delta between original test and retest.
    Provides quantifiable evidence of risk reduction for compliance.
    """

    # Build lookup by finding ID
    original_by_id = {f['id']: f for f in original_findings}
    retest_by_id = {r['finding_id']: r for r in retest_results}

    # Calculate aggregate scores
    original_total_cvss = sum(f['cvss'] for f in original_findings)

    remediated_cvss = 0
    partially_remediated_cvss = 0
    not_remediated_cvss = 0

    finding_details = []

    for finding_id, original in original_by_id.items():
        retest = retest_by_id.get(finding_id, {})
        status = retest.get('status', 'NOT_RETESTED')

        if status == 'REMEDIATED':
            new_cvss = 0.0
            remediated_cvss += original['cvss']
        elif status == 'PARTIALLY_REMEDIATED':
            new_cvss = retest.get('new_cvss', original['cvss'] * 0.5)
            partially_remediated_cvss += new_cvss
        else:
            new_cvss = original['cvss']
            not_remediated_cvss += new_cvss

        finding_details.append({
            'id': finding_id,
            'title': original['title'],
            'original_cvss': original['cvss'],
            'original_severity': original['severity'],
            'status': status,
            'new_cvss': new_cvss,
            'cvss_delta': original['cvss'] - new_cvss,
            'risk_eliminated': status == 'REMEDIATED'
        })

    post_retest_total = partially_remediated_cvss + not_remediated_cvss
    risk_reduction_pct = ((original_total_cvss - post_retest_total) /
                          original_total_cvss * 100) if original_total_cvss > 0 else 0

    return {
        'original_test_date': original_findings[0].get('test_date'),
        'retest_date': retest_results[0].get('retest_date') if retest_results else None,

        'finding_counts': {
            'original': len(original_findings),
            'remediated': sum(1 for r in retest_results if r['status'] == 'REMEDIATED'),
            'partially_remediated': sum(1 for r in retest_results if r['status'] == 'PARTIALLY_REMEDIATED'),
            'not_remediated': sum(1 for r in retest_results if r['status'] == 'NOT_REMEDIATED'),
        },

        'cvss_aggregate': {
            'original_total': round(original_total_cvss, 1),
            'post_retest_total': round(post_retest_total, 1),
            'risk_eliminated': round(original_total_cvss - post_retest_total, 1),
            'risk_reduction_percentage': round(risk_reduction_pct, 1)
        },

        'severity_breakdown': {
            'critical': {
                'original': sum(1 for f in original_findings if f['severity'] == 'CRITICAL'),
                'remaining': sum(1 for r in retest_results
                               if r['status'] != 'REMEDIATED' and
                               original_by_id[r['finding_id']]['severity'] == 'CRITICAL')
            },
            'high': {
                'original': sum(1 for f in original_findings if f['severity'] == 'HIGH'),
                'remaining': sum(1 for r in retest_results
                               if r['status'] != 'REMEDIATED' and
                               original_by_id[r['finding_id']]['severity'] == 'HIGH')
            },
        },

        'findings': finding_details,

        'compliance_summary': {
            'pci_dss_compliant': not_remediated_cvss == 0 and partially_remediated_cvss == 0,
            'soc2_evidence_complete': len(finding_details) == len(original_findings),
            'all_critical_remediated': all(
                r['status'] == 'REMEDIATED'
                for r in retest_results
                if original_by_id[r['finding_id']]['severity'] == 'CRITICAL'
            )
        }
    }

# Example output:
"""
CVSS Delta Report
=================
Original Test Date: 2026-02-15
Retest Date: 2026-03-01

Finding Summary:
  Total original findings: 12
  Fully remediated: 9
  Partially remediated: 2
  Not remediated: 1

CVSS Aggregate:
  Original total CVSS: 67.4
  Post-retest total CVSS: 11.2
  Risk eliminated: 56.2 (83.4%)

Severity Breakdown:
  Critical: 2 original → 0 remaining ✓
  High: 4 original → 1 remaining ✗
  Medium: 6 original → 2 remaining ✗

Compliance Status:
  PCI-DSS: NOT YET COMPLIANT (1 high finding not remediated)
  SOC 2: Evidence package complete for 9 remediated findings
  Critical findings: All remediated ✓
"""

# Mental model for root cause vs symptom:

# Finding: "SQL injection on /api/users/search?name=X"
# Symptom fix: Add input sanitization to this endpoint ← FAILS variation testing
# Root cause fix: Use parameterized queries everywhere ← PASSES variation testing

# Finding: "IDOR on GET /api/orders/{id}"
# Symptom fix: Add ownership check to GET /api/orders/{id} ← FAILS variation
# Root cause fix: Implement ownership-enforced query manager for all Order operations

# Finding: "JWT accepts alg:none"
# Symptom fix: Check for "none" string in algorithm ← FAILS case variation
# Root cause fix: Strict algorithm allowlist ['HS256'] only ← PASSES all variations

# Finding: "hardcoded AWS key in JavaScript bundle"
# Symptom fix: Remove the key from that file ← FAILS if pattern repeated elsewhere
# Root cause fix: Remove all AWS SDK usage from frontend + implement presigned URLs
#                 + add CI/CD secret scanning to prevent recurrence

# Self-test checklist template for common finding types:

SELF_TEST_CHECKLISTS = {
    'IDOR': [
        'Test original endpoint with another user\\'s ID → expect 403/404',
        'Test all HTTP methods on the same endpoint (GET, PUT, DELETE)',
        'Test all sub-resources of the same object',
        'Test the list endpoint for the same resource type',
        'Test with IDs from different tenants (if multi-tenant)',
        'Test with numeric IDs adjacent to own IDs (±1, ±10)',
    ],

    'SQL_INJECTION': [
        'Test with original payload → expect safe response',
        'Test with single quote payload → expect safe response',
        'Test with comment-based payload (--) → expect safe response',
        'Test with UNION-based payload → expect safe response',
        'Test with time-based payload (SLEEP/WAITFOR) → timing should be normal',
        'Test with all input fields on the same endpoint',
        'Test with same input pattern on adjacent endpoints',
    ],

    'CORS_MISCONFIGURATION': [
        'Test with original attacker origin → expect no ACAO header or 403',
        'Test with different attacker domain → expect same result',
        'Test with null origin → expect no reflected null',
        'Test with HTTP variant of trusted domain → expect rejection',
        'Test subdomain variations of trusted domain',
        'Confirm legitimate origins still work (don\\'t break production)',
    ],

    'AUTHENTICATION_BYPASS': [
        'Test original bypass technique → expect 401/403',
        'Test endpoint without any credentials → expect 401',
        'Test with expired credentials → expect 401',
        'Test with invalid token signature → expect 401',
        'Test all HTTP methods without credentials',
        'Test adjacent endpoints in same namespace',
    ],

    'SECRETS_IN_BUNDLE': [
        'Download and search current production bundle → no secrets found',
        'Search all JS files, not just main bundle (chunk files too)',
        'Search source maps if still deployed',
        'Verify secret has been rotated (test old secret doesn\\'t work)',
        'Scan git history for the same secret (ensure it\\'s removed)',
        'Run gitleaks on current codebase',
    ],
}