t0ddc3by/claude-for-customer-success/expansion-business-case

csql_package_created = False
csm_led_proposal_created = False

import html, unicodedata, re
def xml_structural_escape(value: str) -> str:
    """
    Two-layer structural escape for XML/HTML injection defense.
    Layer 1 of 2-layer defense. Returns display-safe string.
    Step 0: html.unescape() — single pass; resolves double-encoding
            (e.g., < → <) and 4-hex-digit entities (<)
    Step 1: NFKC normalization — collapse width variants and compatibility forms
    Step 2: Strip raw < and > individually via two full-string passes
            (no separator inserted between adjacent tag residues)
    Step 3: HTML entity regex — matches &lt; &gt; &#60; &#62; &#x3c; &#x3e;
            and variants with leading zeros: #[xX]0*3[cCeE]
    Step 4: Unicode homoglyphs — explicit iteration over 10 characters:
            U+003C <, U+003E >, U+2039 ‹, U+203A ›, U+27E8 ⟨, U+27E9 ⟩,
            U+3008 〈, U+3009 〉, U+FE64 ﹤, U+FE65 ﹥
    """
    # Step 0: single-pass html.unescape
    value = html.unescape(value)
    # Step 1: NFKC normalization
    value = unicodedata.normalize('NFKC', value)
    # Step 2: strip raw angle brackets individually (no separator)
    value = value.replace('<', '').replace('>', '')
    # Step 3: HTML entity regex (leading zeros, case variants)
    value = re.sub(r'&(?:lt|gt|#(?:60|62|[xX]0*3[cCeE]));', '', value)
    # Step 4: Unicode homoglyph iteration
    for ch in '<>‹›⟨⟩〈〉﹤﹥':
        value = value.replace(ch, '')
    return value
display_account = xml_structural_escape(account_name)  # document output only
safe_account = re.sub(r'[^\w\-]', '_', display_account)  # filesystem ops only
safe_csm_name = xml_structural_escape(csm_name)  # document output only

def normalize_expansion_signals(signals) -> str:
    """
    Normalize expansion_signals to a single escaped string.
    Accepts string or list (max 20 elements, max 500 chars/element).
    Per-element xml_structural_escape applied before join.
    """
    if isinstance(signals, list):
        signals = signals[:20]  # max 20 elements
        escaped = [xml_structural_escape(str(s)[:500]) for s in signals]
        return '; '.join(escaped)
    return xml_structural_escape(str(signals))
safe_signals = normalize_expansion_signals(expansion_signals)

safe_outcome_evidence = xml_structural_escape(outcome_evidence)
safe_expansion_product = xml_structural_escape(expansion_product)
safe_timeline = xml_structural_escape(timeline) if timeline else None
safe_expansion_amount = xml_structural_escape(expansion_amount) if expansion_amount else None
if csql_context:
    safe_csql = {k: xml_structural_escape(str(v)) for k, v in csql_context.items()}
else:
    safe_csql = {}

INJECTION_PATTERNS = [
    # Patterns 0-9: content injection — matched case-insensitively via lower_value
    r'\bignore\b.{0,30}\b(?:previous|above|prior|instructions?|directives?|rules?|constraints?)\b',
    r'\bdisregard\b.{0,30}\b(?:previous|above|prior|instructions?|directives?|rules?|constraints?)\b',
    r'\bforget\b.{0,30}\b(?:previous|above|prior|instructions?|directives?|rules?|constraints?)\b',
    r'\bact\s+as\b.{0,50}\b(?:different|new|another|alternate|alternative)\b',
    r'\b(?:you\s+are|you\'re)\s+(?:now\s+)?(?:a\s+|an\s+)(?!CSM|customer)',
    r'\b(?:pretend|roleplay|role-play|simulate)\b.{0,30}\b(?:you\s+are|you\'re|as\b)',
    r'\bdo\s+not\b.{0,30}\b(?:follow|apply|use|respect|enforce)\b.{0,30}\b(?:rules?|constraints?|instructions?|guidelines?)\b',
    r'\b(?:reveal|expose|output|print|display|show)\b.{0,30}\b(?:system\s+prompt|instructions?|directives?|rules?)\b',
    r'\boverride\w*\b.{0,30}\b(?:instructions?|rules?|constraints?|directives?|guidelines?)\b',
    r'\b(?:inject|insertion|payload)\b.{0,30}\b(?:prompt|instruction|directive|command)\b',
    # Patterns 10-12: structural LLM-format injection (re.IGNORECASE on original value)
    r'\[(?:INST|SYS|SYSTEM|END)\]',
    r'(?:###\s*)?(?:instruction|system|user|assistant)\s*:\s',
    r'\bhuman\s*:\s|\bassistant\s*:\s',
]
def scan_for_injection(value: str) -> bool:
    """
    Returns True if injection attempt detected. Patterns 0-9 match against
    lowercased value for case-insensitive coverage. Patterns 10-12 use
    re.IGNORECASE on the original value (preserves delimiter detection).
    """
    lower_value = value.lower()
    for i, pattern in enumerate(INJECTION_PATTERNS):
        target = lower_value if i < 10 else value
        flags = re.IGNORECASE if i >= 10 else 0
        if re.search(pattern, target, flags):
            return True
    return False
# Scan all escaped inputs
inputs_to_scan = [
    safe_outcome_evidence, safe_expansion_product, safe_signals,
    safe_csm_name, safe_timeline or '', safe_expansion_amount or '',
] + list(safe_csql.values())
for val in inputs_to_scan:
    if scan_for_injection(val):
        raise ValueError(
            "Input rejected: potential prompt injection detected. "
            "Please review your inputs and remove any instruction-like content."
        )

constraint_warnings = []
constraint_errors = []
# C-3: CSQL minimum inputs (BLOCKING)
if mode == 'csql':
    required_csql_fields = ['economic_buyer', 'identified_pain', 'champion']
    missing = [f for f in required_csql_fields if not safe_csql.get(f)]
    if missing:
        constraint_errors.append(
            f"C-3 ERROR: CSQL mode requires {', '.join(missing)}. "
            f"Provide these fields in csql_context before proceeding."
        )
# C-4: Mutual exclusivity (BLOCKING)
if mode == 'csm-led' and csm_led_proposal_created:
    constraint_errors.append(
        "C-4 ERROR: A csm-led proposal has already been generated for this "
        "invocation. Start a new skill session to generate an additional proposal."
    )
if mode == 'csql' and csql_package_created:
    constraint_errors.append(
        "C-4 ERROR: A CSQL package has already been generated for this "
        "invocation. Start a new skill session to generate an additional package."
    )
if constraint_errors:
    raise ValueError('\n'.join(constraint_errors))
# C-1: Health score (non-blocking)
if health_score is not None and health_score < 70:
    constraint_warnings.append(
        f"⚠️ C-1: Account health score is {health_score}/100 (below 70). "
        "Review health trajectory before presenting expansion to customer."
    )
# C-2: Utilization rate (non-blocking)
if utilization_rate is not None and utilization_rate < 60:
    constraint_warnings.append(
        f"⚠️ C-2: Current utilization is {utilization_rate}% (below 60%). "
        "Consider whether expansion is premature given current adoption level."
    )
# C-5: Economic buyer placeholder (non-blocking)
economic_buyer = safe_csql.get('economic_buyer', '') if safe_csql else ''
if not economic_buyer:
    economic_buyer = '[Economic Buyer — confirm with AE]'
    constraint_warnings.append(
        "⚠️ C-5: Economic buyer not identified. Placeholder inserted in output. "
        "Confirm with AE before customer-facing use."
    )