Triage Issue

#!/usr/bin/env python3
"""
Detect prompt injection attempts and non-English content in GitHub issues.

This script performs two security checks:
1. Language check: Reject non-English issues
2. Prompt injection check: Detect malicious patterns in English text

Usage:
  detect_prompt_injection.py <issue-json-file> [comments-json-file]

  issue-json-file    - GitHub issue JSON (single object with title/body)
  comments-json-file - Optional GitHub comments JSON (array of comment objects)
                       When provided, all comment bodies are checked for injection.
                       Language check is skipped for comments (issue already passed).

Exit codes:
  0 - Safe to proceed (English + no injection detected)
  1 - REJECT: Non-English content or injection detected
  2 - Error reading input
"""

import json
import re
import sys
from typing import List, Tuple


def is_english(text: str) -> Tuple[bool, float]:
    """
    Check if text is primarily English.

    Strategy:
    1. Reject text where a significant fraction of alphabetic characters are
       non-ASCII (covers Cyrillic, CJK, Arabic, Hebrew, Thai, Hangul, etc.).
    2. Also reject text that contains accented Latin characters common in
       Romance/Germanic languages (é, ñ, ö, ç, etc.).

    Args:
        text: Text to check

    Returns:
        (is_english, ascii_ratio)
    """
    if not text or len(text.strip()) < 20:
        return True, 1.0  # Too short to determine, assume OK

    total_alpha = sum(1 for c in text if c.isalpha())
    if total_alpha == 0:
        return True, 1.0

    ascii_alpha = sum(1 for c in text if c.isascii() and c.isalpha())
    ratio = ascii_alpha / total_alpha

    # If more than 20% of alphabetic characters are non-ASCII, treat as
    # non-English. This catches Cyrillic, CJK, Arabic, Hebrew, Thai,
    # Hangul, Devanagari, and any other non-Latin script.
    if ratio < 0.80:
        return False, ratio

    # For text that is mostly ASCII, also reject known non-Latin script
    # characters that could appear as a small minority (e.g. a single
    # Cyrillic word embedded in otherwise ASCII text).
    NON_LATIN_RANGES = [
        (0x0400, 0x04FF),  # Cyrillic
        (0x0500, 0x052F),  # Cyrillic Supplement
        (0x0600, 0x06FF),  # Arabic
        (0x0590, 0x05FF),  # Hebrew
        (0x0E00, 0x0E7F),  # Thai
        (0x3040, 0x309F),  # Hiragana
        (0x30A0, 0x30FF),  # Katakana
        (0x4E00, 0x9FFF),  # CJK Unified Ideographs
        (0xAC00, 0xD7AF),  # Hangul Syllables
        (0x0900, 0x097F),  # Devanagari
        (0x0980, 0x09FF),  # Bengali
        (0x0A80, 0x0AFF),  # Gujarati
        (0x0C00, 0x0C7F),  # Telugu
        (0x0B80, 0x0BFF),  # Tamil
    ]

    def is_non_latin(c: str) -> bool:
        cp = ord(c)
        return any(start <= cp <= end for start, end in NON_LATIN_RANGES)

    non_latin_count = sum(1 for c in text if is_non_latin(c))
    if non_latin_count > 3:
        return False, ratio

    # Common accented characters in Romance and Germanic languages
    # These rarely appear in English bug reports
    NON_ENGLISH_CHARS = set('áéíóúàèìòùâêîôûäëïöüãõñçßø')
    text_lower = text.lower()
    has_non_english = any(c in NON_ENGLISH_CHARS for c in text_lower)

    if has_non_english:
        return False, ratio

    return True, 1.0


# ============================================================================
# PROMPT INJECTION PATTERNS (English only)
# ============================================================================
# High-confidence patterns that indicate malicious intent

INJECTION_PATTERNS = [
    # System override tags and markers (10 points each)
    (r"<\s*system[_\s-]*(override|message|prompt|instruction)", 10, "System tag injection"),
    (r"\[system[\s_-]*(override|message|prompt)", 10, "System marker injection"),
    (r"<!--\s*(claude|system|admin|override):", 10, "HTML comment injection"),

    # Instruction override attempts (8 points)
    (r"\b(ignore|disregard|forget)\s+(all\s+)?(previous|prior|above)\s+(instructions?|prompts?|rules?)", 8, "Instruction override"),

    # Prompt extraction (8 points)
    (r"\b(show|reveal|display|output|print)\s+(your\s+)?(system\s+)?(prompt|instructions?)", 8, "Prompt extraction attempt"),
    (r"\bwhat\s+(is|are)\s+your\s+(system\s+)?(prompt|instructions?)", 8, "Prompt extraction question"),

    # Role manipulation (8 points)
    (r"\byou\s+are\s+now\s+(in\s+)?((an?\s+)?(admin|developer|debug|system|root))", 8, "Role manipulation"),
    (r"\b(admin|developer|system)[\s_-]mode", 8, "Mode manipulation"),

    # Sensitive file paths (10 points) - legitimate issues rarely reference these
    (r"(~/\.aws/|~/\.ssh/|/root/|/etc/passwd|/etc/shadow)", 10, "System credentials path"),
    (r"(\.aws/credentials|id_rsa|\.ssh/id_)", 10, "Credentials file reference"),

    # Environment variable exfiltration (8 points)
    (r"\$(aws_secret|aws_access|github_token|anthropic_api|api_key|secret_key)", 8, "Sensitive env var reference"),
    (r"process\.env\.(secret|token|password|api)", 7, "Process.env access"),

    # Command execution attempts (7 points)
    (r"`\s*(env|printenv|cat\s+[~/]|grep\s+secret)", 7, "Suspicious command in code block"),
    (r"\b(run|execute).{0,10}(command|script|bash)", 6, "Command execution request"),
    (r"running\s+(this|the)\s+command:\s*`", 6, "Command execution with backticks"),

    # Credential harvesting (7 points)
    (r"\bsearch\s+for.{0,10}(api.?keys?|tokens?|secrets?|passwords?)", 7, "Credential search request"),
    (r"\b(read|check|access).{0,30}(credentials|\.env|api.?key)", 6, "Credentials access request"),

    # False authorization (6 points)
    (r"\b(i\s+am|i'm|user\s+is).{0,15}(authorized|approved)", 6, "False authorization claim"),
    (r"(verification|admin|override).?code:?\s*[a-z][a-z0-9]{2,}[-_][a-z0-9]{3,}", 6, "Fake verification code"),

    # Chain-of-thought manipulation (6 points)
    (r"\b(actually|wait),?\s+(before|first|instead)", 6, "Instruction redirect"),
    (r"let\s+me\s+think.{0,20}what\s+you\s+should\s+(really|actually)", 6, "CoT manipulation"),

    # Script/iframe injection (10 points)
    (r"<\s*script[^>]*\s(src|onerror|onload)\s*=", 10, "Script tag injection"),
    (r"<\s*iframe[^>]*src\s*=", 10, "Iframe injection"),
]


def check_injection(text: str, threshold: int = 8) -> Tuple[bool, int, List[str]]:
    """
    Check English text for prompt injection patterns.

    Args:
        text: Text to check (assumed to be English)
        threshold: Minimum score to trigger detection (default: 8)

    Returns:
        (is_injection_detected, total_score, list_of_matches)
    """
    if not text:
        return False, 0, []

    total_score = 0
    matches = []

    normalized = text.lower()

    for pattern, score, description in INJECTION_PATTERNS:
        if re.search(pattern, normalized, re.MULTILINE):
            total_score += score
            matches.append(f"  - {description} (+{score} points)")

    is_injection = total_score >= threshold
    return is_injection, total_score, matches


def analyze_issue(issue_data: dict) -> Tuple[bool, str, List[str]]:
    """
    Analyze issue for both language and prompt injection.

    Returns:
        (should_reject, reason, details)
        - should_reject: True if triage should abort
        - reason: "non-english", "injection", or None
        - details: List of strings describing the detection
    """
    title = issue_data.get("title", "")
    body = issue_data.get("body", "")

    # Combine title and body for checking
    combined_text = f"{title}\n\n{body}"

    # Check 1: Language detection
    is_eng, ratio = is_english(combined_text)

    if not is_eng:
        details = [
            f"Language check failed: non-English characters detected ({ratio:.1%} ASCII alphabetic)",
            "",
            "This triage system only processes English language issues.",
            "Please submit issues in English for automated triage.",
        ]
        return True, "non-english", details

    # Check 2: Prompt injection detection
    is_injection, score, matches = check_injection(combined_text)

    if is_injection:
        details = [
            f"Prompt injection detected (score: {score} points)",
            "",
            "Matched patterns:",
        ] + matches
        return True, "injection", details

    # All checks passed
    return False, None, ["Language: English ✓", "Injection check: Passed ✓"]


def analyze_comments(comments_data: list) -> Tuple[bool, str, List[str]]:
    """
    Check issue comments for prompt injection. Language check is skipped
    because the issue body already passed; comments are checked for injection only.

    Args:
        comments_data: List of GitHub comment objects (each has a "body" field)

    Returns:
        (should_reject, reason, details)
    """
    for i, comment in enumerate(comments_data):
        if not isinstance(comment, dict):
            continue
        body = comment.get("body") or ""
        if not body:
            continue

        is_injection, score, matches = check_injection(body)
        if is_injection:
            author = comment.get("user", {}).get("login", "unknown")
            details = [
                f"Prompt injection detected in comment #{i + 1} by @{author} (score: {score} points)",
                "",
                "Matched patterns:",
            ] + matches
            return True, "injection", details

    return False, None, ["Comments injection check: Passed ✓"]


def main():
    if len(sys.argv) not in (2, 3):
        print("Usage: detect_prompt_injection.py <issue-json-file> [comments-json-file]", file=sys.stderr)
        sys.exit(2)

    json_file = sys.argv[1]

    try:
        with open(json_file, 'r', encoding='utf-8') as f:
            issue_data = json.load(f)
    except Exception as e:
        print(f"Error reading issue JSON file: {e}", file=sys.stderr)
        sys.exit(2)

    should_reject, reason, details = analyze_issue(issue_data)

    if should_reject:
        print("=" * 60)
        if reason == "non-english":
            print("REJECTED: Non-English content detected")
        elif reason == "injection":
            print("REJECTED: Prompt injection attempt detected")
        print("=" * 60)
        print()
        for line in details:
            print(line)
        print()
        sys.exit(1)

    # Check comments if provided
    if len(sys.argv) == 3:
        comments_file = sys.argv[2]
        try:
            with open(comments_file, 'r', encoding='utf-8') as f:
                comments_data = json.load(f)
        except Exception as e:
            print(f"Error reading comments JSON file: {e}", file=sys.stderr)
            sys.exit(2)

        if not isinstance(comments_data, list):
            print("Error: comments JSON must be an array", file=sys.stderr)
            sys.exit(2)

        should_reject, reason, comment_details = analyze_comments(comments_data)
        details.extend(comment_details)

        if should_reject:
            print("=" * 60)
            print("REJECTED: Prompt injection attempt detected")
            print("=" * 60)
            print()
            for line in comment_details:
                print(line)
            print()
            sys.exit(1)

    print("Security checks passed")
    for line in details:
        print(line)
    sys.exit(0)


if __name__ == "__main__":
    main()

"""
Parse GitHub API JSON (single issue or search/issues) and print a concise summary.
Reads from stdin if no argument, else from the file path given as first argument.
Used by the triage-issue skill in CI so the AI does not need inline python3 -c in Bash.
"""
import json
import sys


def _sanitize_title(title: str) -> str:
    """One line, no leading/trailing whitespace, newlines replaced with space."""
    if not title:
        return ""
    return " ".join(str(title).split())


def _format_single_issue(data: dict) -> None:
    num = data.get("number")
    title = _sanitize_title(data.get("title", ""))
    state = data.get("state", "")
    print(f"#{num} {title} {state}")
    labels = data.get("labels", [])
    if labels:
        names = [l.get("name", "") for l in labels if isinstance(l, dict)]
        print(f"Labels: {', '.join(names)}")
    body = data.get("body") or ""
    if body:
        snippet = body[:200].replace("\n", " ")
        if len(body) > 200:
            snippet += "..."
        print(f"Body: {snippet}")


def _format_search_items(data: dict) -> None:
    items = data.get("items", [])
    for i in items:
        if not isinstance(i, dict):
            continue
        num = i.get("number", "")
        title = _sanitize_title(i.get("title", ""))
        state = i.get("state", "")
        print(f"{num} {title} {state}")


def main() -> None:
    if len(sys.argv) > 1:
        path = sys.argv[1]
        try:
            with open(path, encoding="utf-8") as f:
                data = json.load(f)
        except (OSError, json.JSONDecodeError) as e:
            print(f"parse_gh_issues: {e}", file=sys.stderr)
            sys.exit(1)
    else:
        try:
            data = json.load(sys.stdin)
        except json.JSONDecodeError as e:
            print(f"parse_gh_issues: {e}", file=sys.stderr)
            sys.exit(1)

    if not isinstance(data, dict):
        print("parse_gh_issues: expected a JSON object", file=sys.stderr)
        sys.exit(1)

    if "items" in data:
        _format_search_items(data)
    elif "number" in data:
        _format_single_issue(data)
    else:
        print("parse_gh_issues: expected 'items' (search) or 'number' (single issue)", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

import json, os, re, sys, urllib.error, urllib.request, urllib.parse

TIMEOUT_SECONDS = 30
IDENTIFIER_PATTERN = re.compile(r"^[A-Z]+-\d+$")
# In CI only the workspace (cwd) is writable; /tmp/ is allowed for local runs
ALLOWED_REPORT_PREFIXES = ("/tmp/", os.path.abspath(os.getcwd()) + os.sep)


def _report_path_allowed(path: str) -> bool:
    abs_path = os.path.abspath(path)
    return any(abs_path.startswith(p) for p in ALLOWED_REPORT_PREFIXES)


def graphql(token, query, variables=None):
    payload = json.dumps({"query": query, **({"variables": variables} if variables else {})}).encode()
    req = urllib.request.Request(
        "https://api.linear.app/graphql",
        data=payload,
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"},
    )
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")
        print(f"Linear API error {e.code}: {body}")
        sys.exit(1)
    except urllib.error.URLError as e:
        print(f"Linear API request failed: {e.reason}")
        sys.exit(1)


# --- Inputs ---
identifier = sys.argv[1]       # e.g. "JS-1669"
report_path = sys.argv[2]      # e.g. "triage_report.md" (repo root; in CI use repo root only)

if not IDENTIFIER_PATTERN.match(identifier):
    print(f"Invalid identifier format: {identifier}")
    sys.exit(1)

if not _report_path_allowed(report_path):
    print(
        f"Report path must be under current working directory ({os.getcwd()}) or /tmp/. In CI use repo root, e.g. triage_report.md"
    )
    sys.exit(1)

client_id = os.environ["LINEAR_CLIENT_ID"]
client_secret = os.environ["LINEAR_CLIENT_SECRET"]

# --- Obtain access token ---
token_data = urllib.parse.urlencode({
    "grant_type": "client_credentials",
    "client_id": client_id,
    "client_secret": client_secret,
    "scope": "issues:create,read,comments:create",
}).encode()
req = urllib.request.Request("https://api.linear.app/oauth/token", data=token_data,
    headers={"Content-Type": "application/x-www-form-urlencoded"})
try:
    with urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS) as resp:
        token = json.loads(resp.read()).get("access_token", "")
except (urllib.error.HTTPError, urllib.error.URLError) as e:
    print(f"Failed to obtain Linear access token: {e}")
    sys.exit(1)
if not token:
    print("Failed to obtain Linear access token")
    sys.exit(1)

# --- Fetch issue UUID ---
data = graphql(token,
    "query GetIssue($id: String!) { issue(id: $id) { id identifier url } }",
    {"id": identifier},
)
issue = data.get("data", {}).get("issue")
if not issue:
    print(f"Linear issue {identifier} not found")
    sys.exit(1)
issue_id = issue["id"]

# --- Check for existing triage comment (idempotency) ---
data = graphql(token,
    "query GetComments($id: String!) { issue(id: $id) { comments { nodes { body } } } }",
    {"id": identifier},
)
comments = data.get("data", {}).get("issue", {}).get("comments", {}).get("nodes", [])
for c in comments:
    if c.get("body", "").startswith("## Issue Triage:"):
        print(f"Triage comment already exists on {identifier}, skipping")
        sys.exit(0)

# --- Post comment ---
with open(report_path) as f:
    body = f.read()
data = graphql(token,
    "mutation CommentCreate($input: CommentCreateInput!) { commentCreate(input: $input) { success comment { id } } }",
    {"input": {"issueId": issue_id, "body": body}},
)
if data.get("data", {}).get("commentCreate", {}).get("success"):
    print(f"Triage comment posted on {identifier}: {issue['url']}")
else:
    print(f"Failed to post triage comment: {json.dumps(data)}")
    sys.exit(1)

#!/usr/bin/env python3
"""
Read Claude Code execution output JSON and write duration, cost, and status
to stdout as Markdown for GitHub Actions job summary (GITHUB_STEP_SUMMARY).

Usage:
  python3 write_job_summary.py <path-to-claude-execution-output.json>

The execution file is written by anthropics/claude-code-action as a single
JSON array of messages (JSON.stringify(messages, null, 2)) at
$RUNNER_TEMP/claude-execution-output.json. We also support NDJSON (one
object per line). Uses the last object with type "result" for metrics.

Job summary has a ~1MB limit; raw JSON is truncated if needed to avoid job abort.
"""

import json
import sys

# Stay under GITHUB_STEP_SUMMARY ~1MB limit; leave room for the table and text
MAX_RAW_BYTES = 800_000


def _append_raw_json_section(content: str, lines: list[str]) -> None:
    """Append a 'Full execution output' json block to lines, with truncation and fence escaping."""
    raw = content.strip()
    encoded = raw.encode("utf-8")
    if len(encoded) > MAX_RAW_BYTES:
        raw = encoded[:MAX_RAW_BYTES].decode("utf-8", errors="replace") + "\n\n... (truncated due to job summary size limit)"
    raw = raw.replace("```", "`\u200b``")
    lines.extend(["", "### Full execution output", "", "```json", raw, "```"])


def main() -> int:
    if len(sys.argv) < 2:
        print("Usage: write_job_summary.py <execution-output.json>", file=sys.stderr)
        return 1

    path = sys.argv[1]
    try:
        with open(path, encoding="utf-8") as f:
            content = f.read()
    except OSError as e:
        msg = f"## Claude Triage Run\n\nCould not read execution output: {e}"
        print(msg, file=sys.stderr)
        print(msg)  # Also to stdout so job summary shows something
        return 1

    # Support single JSON or NDJSON (one object per line)
    results = []
    for line in content.strip().splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            obj = json.loads(line)
            if isinstance(obj, dict) and obj.get("type") == "result":
                results.append(obj)
            elif isinstance(obj, list):
                for item in obj:
                    if isinstance(item, dict) and item.get("type") == "result":
                        results.append(item)
        except json.JSONDecodeError:
            continue

    if not results:
        # Try parsing whole content as single JSON (object or array)
        try:
            obj = json.loads(content)
            if isinstance(obj, dict) and obj.get("type") == "result":
                results = [obj]
            elif isinstance(obj, list):
                for item in obj:
                    if isinstance(item, dict) and item.get("type") == "result":
                        results.append(item)
        except json.JSONDecodeError:
            pass

    if not results:
        no_result_lines = ["## Claude Triage Run", "", "No execution result found in output."]
        _append_raw_json_section(content, no_result_lines)
        print("\n".join(no_result_lines))
        return 0

    last = results[-1]
    duration_ms = last.get("duration_ms")
    num_turns = last.get("num_turns")
    total_cost = last.get("total_cost_usd")
    subtype = last.get("subtype", "")

    cost_str = f"${total_cost:.4f} USD" if isinstance(total_cost, (int, float)) else "n/a"
    lines = [
        "## Claude Triage Run",
        "",
        "| Metric | Value |",
        "|--------|-------|",
        f"| Duration | {duration_ms if duration_ms is not None else 'n/a'} ms |",
        f"| Turns | {num_turns if num_turns is not None else 'n/a'} |",
        f"| Cost (USD) | {cost_str} |",
    ]
    if subtype == "error_max_turns":
        lines.extend([
            "",
            "⚠️ **Run stopped:** maximum turns reached. Consider increasing `max-turns` in the workflow or simplifying the issue scope.",
        ])
    elif subtype and subtype != "success":
        lines.extend([
            "",
            f"Result: `{subtype}`",
        ])

    _append_raw_json_section(content, lines)

    print("\n".join(lines))
    return 0


if __name__ == "__main__":
    sys.exit(main())