config.yaml.example
# Cortex Code Skill Configuration Example
#
# Copy this file to config.yaml in the skill's install directory and customize as needed.
# Install directory varies by agent:
# Claude Code: ~/.claude/skills/cortex-code/config.yaml
# Cursor: ~/.cursor/skills/cortex-code/config.yaml
# Windsurf: ~/.windsurf/skills/cortex-code/config.yaml
# VSCode: see agent docs for skills directory location
#
# For detailed documentation, see:
# - SECURITY.md - Security features and policies
# - SECURITY_GUIDE.md - Deployment best practices
# - README.md - General usage guide
# ==============================================================================
# SECURITY CONFIGURATION
# ==============================================================================
security:
# ----------------------------------------------------------------------------
# APPROVAL MODE (MOST IMPORTANT SETTING)
# ----------------------------------------------------------------------------
# Controls how tool execution is approved before running Cortex Code.
#
# Options:
# "prompt" - Show approval prompt before execution (DEFAULT, MOST SECURE)
# User must review and approve predicted tools.
# Best for: Interactive use, security-sensitive environments
#
# "auto" - Auto-approve all operations
# Requires mandatory audit logging.
# Best for: Trusted environments, automated workflows
#
# "envelope_only" - No tool prediction, rely on envelope blocklist only
# Faster than "auto", still requires audit logging.
# Best for: Trust Cortex Code's envelope enforcement
#
# SECURITY: Default is "prompt" for maximum security.
#
approval_mode: "prompt"
# ----------------------------------------------------------------------------
# TOOL PREDICTION (for "prompt" mode)
# ----------------------------------------------------------------------------
# Confidence threshold for tool prediction (0.0 to 1.0)
# If prediction confidence is below this threshold, a warning is shown.
#
# Default: 0.7 (70% confidence)
# Lower values = more lenient, fewer warnings
# Higher values = stricter, more warnings
#
tool_prediction_confidence_threshold: 0.7
# ----------------------------------------------------------------------------
# AUDIT LOGGING (mandatory for "auto" and "envelope_only" modes)
# ----------------------------------------------------------------------------
# Structured JSONL logging of all executions.
# Format: One JSON object per line (machine-readable)
#
# Log location (supports ~/ and environment variables)
# audit_log defaults to audit.log in the skill install directory (set automatically)
# Override: audit_log_path: "~/.your-agent/skills/cortex-code/audit.log"
audit_log_path: "~/.cache/cortex-skill/audit.log"
# Log rotation size (e.g., "10MB", "50MB", "100MB")
# When log exceeds this size, it's rotated to audit.log.1, audit.log.2, etc.
audit_log_rotation: "10MB"
# Log retention in days
# Logs older than this are deleted during rotation
audit_log_retention: 30
# ----------------------------------------------------------------------------
# PROMPT SANITIZATION
# ----------------------------------------------------------------------------
# Remove PII (emails, phone numbers, SSN, credit cards) and detect injection
# attempts before processing prompts.
#
# SECURITY: Enabled by default. Disable only if you trust all input sources.
#
sanitize_conversation_history: true
# ----------------------------------------------------------------------------
# SECURE CACHING
# ----------------------------------------------------------------------------
# Cache directory for Cortex capabilities and other temporary data.
# Uses SHA256 fingerprint validation for integrity.
#
# Default: ~/.cache/cortex-skill
#
cache_dir: "~/.cache/cortex-skill"
# Cache TTL (time-to-live) in seconds
# Default: 86400 (24 hours)
cache_ttl: 86400
# ----------------------------------------------------------------------------
# CREDENTIAL FILE PROTECTION
# ----------------------------------------------------------------------------
# Blocks routing when prompts contain paths matching these patterns.
# Prevents accidental exposure of sensitive credential files.
#
# Pattern syntax:
# - ~/ = user home directory
# - ** = any subdirectories
# - * = any characters
#
# SECURITY: Add patterns for your organization's credential files.
#
credential_file_allowlist:
# SSH keys
- "~/.ssh/**"
# Cloud provider credentials
- "~/.aws/credentials"
- "~/.aws/config"
- "~/.gcp/**"
- "~/.azure/**"
# Snowflake credentials
- "~/.snowflake/**"
# Environment files
- "**/.env"
- "**/.env.*"
# Generic credential files
- "**/credentials.json"
- "**/credentials.yaml"
- "**/secrets.json"
- "**/secrets.yaml"
# Private keys
- "**/*.pem"
- "**/*.key"
- "**/*_key"
- "**/*-key"
# Language-specific
- "**/.npmrc"
- "**/.pypirc"
- "**/.netrc"
# ----------------------------------------------------------------------------
# SECURITY ENVELOPES
# ----------------------------------------------------------------------------
# Which security envelopes are allowed for execution.
# Envelopes control which tools Cortex Code can use.
#
# Options:
# "RO" - Read-only operations (queries, reads)
# "RW" - Read-write operations (queries, writes, creates)
# "RESEARCH" - Exploratory work with web access
# "DEPLOY" - Deployment operations; destructive shell commands remain blocked
#
# SECURITY: Limit envelopes to your operational needs.
# ENTERPRISE: Consider allowing only RO/RW, require approval for DEPLOY.
#
allowed_envelopes:
- "RO"
- "RW"
- "RESEARCH"
- "DEPLOY"
# ==============================================================================
# EXAMPLE CONFIGURATIONS BY DEPLOYMENT TYPE
# ==============================================================================
# Uncomment the section below that matches your deployment model
# ------------------------------------------------------------------------------
# PERSONAL USE (Individual Developer)
# ------------------------------------------------------------------------------
# Recommended: Secure mode with optional audit logging
#
# security:
# approval_mode: "prompt"
# sanitize_conversation_history: true
# # audit_log defaults to audit.log in the skill install directory (set automatically)
# Override: audit_log_path: "~/.your-agent/skills/cortex-code/audit.log"
audit_log_path: "~/.cache/cortex-skill/audit.log"
# credential_file_allowlist:
# - "~/.ssh/**"
# - "~/.aws/credentials"
# - "~/.snowflake/**"
# - "**/.env"
# ------------------------------------------------------------------------------
# TEAM DEPLOYMENT (5-50 developers)
# ------------------------------------------------------------------------------
# Recommended: Secure mode with mandatory audit logging
# NOTE: Use organization policy file for team-wide enforcement
#
# security:
# approval_mode: "prompt"
# # audit_log defaults to audit.log in the skill install directory (set automatically)
# Override: audit_log_path: "~/.your-agent/skills/cortex-code/audit.log"
audit_log_path: "~/.cache/cortex-skill/audit.log"
# audit_log_retention: 90 # 90 days for team audit
# sanitize_conversation_history: true
# allowed_envelopes:
# - "RO"
# - "RW"
# # RESEARCH and DEPLOY disabled for team safety
# ------------------------------------------------------------------------------
# ENTERPRISE DEPLOYMENT (50+ developers)
# ------------------------------------------------------------------------------
# Recommended: Use organization policy file instead of user config
# Location: ~/.snowflake/cortex/claude-skill-policy.yaml
#
# Organization policy overrides user configuration.
# See SECURITY_GUIDE.md for enterprise deployment details.
#
# security:
# approval_mode: "prompt" # Enforced, no exceptions
# audit_log_path: "/var/log/cortex-skill/audit.log"
# audit_log_retention: 365 # 1 year for compliance
# sanitize_conversation_history: true
# tool_prediction_confidence_threshold: 0.8 # Stricter for enterprise
# allowed_envelopes:
# - "RO" # Only read-only by default
# ------------------------------------------------------------------------------
# AUTO-APPROVAL MODE
# ------------------------------------------------------------------------------
# Use this for auto-approval behavior with audit logging.
#
# security:
# approval_mode: "auto"
# # audit_log defaults to audit.log in the skill install directory (set automatically)
# Override: audit_log_path: "~/.your-agent/skills/cortex-code/audit.log"
audit_log_path: "~/.cache/cortex-skill/audit.log"
# audit_log_rotation: "10MB"
# audit_log_retention: 30
# sanitize_conversation_history: true
# ==============================================================================
# ENVIRONMENT VARIABLE OVERRIDES
# ==============================================================================
#
# You can override configuration via environment variables:
#
# CORTEX_SKILL_CONFIG=/path/to/config.yaml
# Override default config path
#
#
# Example:
# export CORTEX_SKILL_CONFIG=~/.config/cortex-skill/config.yaml
# ==============================================================================
# ORGANIZATION POLICY (for teams/enterprises)
# ==============================================================================
#
# Create organization policy file at:
# ~/.snowflake/cortex/claude-skill-policy.yaml
#
# Organization policy overrides user configuration.
# Deploy via configuration management (Ansible, Puppet, Chef).
#
# Example organization policy:
#
# security:
# approval_mode: "prompt" # Enforced for all users
# # audit_log defaults to audit.log in the skill install directory (set automatically)
# Override: audit_log_path: "~/.your-agent/skills/cortex-code/audit.log"
audit_log_path: "~/.cache/cortex-skill/audit.log"
# sanitize_conversation_history: true
# credential_file_allowlist:
# - "~/.ssh/**"
# - "~/.aws/**"
# - "~/.snowflake/**"
# - "**/.env*"
# allowed_envelopes:
# - "RO"
# - "RW"
# ==============================================================================
# TROUBLESHOOTING
# ==============================================================================
#
# Issue: Approval prompts not appearing
# Solution: Check approval_mode is "prompt" and org policy isn't overriding
#
# Issue: Audit logs not created
# Solution: Ensure log directory exists and has correct permissions (0700)
#
# Issue: All prompts blocked
# Solution: Review credential_file_allowlist patterns, may be too broad
#
# Issue: Cache errors
# Solution: Clear cache directory: rm -rf ~/.cache/cortex-skill/*
#
# For more troubleshooting, see:
# - SECURITY_GUIDE.md - Security configuration help
# ==============================================================================
# ADDITIONAL RESOURCES
# ==============================================================================
#
# Documentation:
# - README.md - General usage and features
# - SECURITY.md - Security policy and threat model
# - SECURITY_GUIDE.md - Deployment best practices
#
# Support:
# - GitHub Issues: https://github.com/Snowflake-Labs/subagent-cortex-code/issues
# - Security: security@snowflake.com
cortex-snowflake-routing.mdc
---
description: Route Snowflake queries to the cortex-code skill for specialized Snowflake expertise via Cortex Code CLI
globs:
alwaysApply: true
---
# Snowflake Query Routing
When the user asks about Snowflake, databases, warehouses, Cortex, or SQL queries, invoke the cortex-code skill with conversation context.
/cortex-code [user's question with relevant context]
## Detection Keywords
Invoke `/cortex-code` when user mentions:
- Snowflake, warehouse, database, schema, table, view
- SQL, query, SELECT, data quality, data analysis
- Cortex Search, Cortex Analyst, Cortex AI
- Snowpark, dynamic tables, streams, tasks
- "how many databases", "show me", "query", "check data"
## How to Invoke
1. **Detect Snowflake query**
2. **Include context**: If there were previous Snowflake-related exchanges in this conversation, include that context
3. **Invoke skill**: Call `/cortex-code` with enriched query
4. **Display results**: Show output from Cortex Code agent
## Examples
**Standalone query:**
User: "How many databases do I have in Snowflake?"
You: /cortex-code How many databases do I have in Snowflake?
**Query with context:**
User: "Which databases have stock data?" → [answered: DB_STOCK, FINANCE__ECONOMICS]
User: "Show me the schema for the main table"
You: /cortex-code User previously identified databases with stock data: DB_STOCK, FINANCE__ECONOMICS. Show me the schema for the main table in DB_STOCK.
## Important
- Do NOT answer Snowflake questions yourself
- ALWAYS invoke `/cortex-code` skill
- Include prior conversation context when relevant
- The skill handles: Cortex routing, SQL execution, formatting
## Non-Snowflake Queries
Handle normally without skill:
- General programming questions
- Local file operations
- Git operations
- Non-Snowflake databases (PostgreSQL, MySQL, etc.)
scripts/discover_cortex.py
#!/usr/bin/env python3
"""
Discovers Cortex Code capabilities by listing skills and parsing their metadata.
Caches results for the current CodingAgent session.
"""
import argparse
import json
import subprocess
import sys
from pathlib import Path
import re
# Add parent directory to path for security imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from security.cache_manager import CacheManager
from security.config_manager import ConfigManager
def run_command(cmd):
"""Run a command and return output."""
try:
result = subprocess.run(
cmd,
shell=False,
capture_output=True,
text=True,
timeout=10
)
return result.stdout, result.stderr, result.returncode
except subprocess.TimeoutExpired:
return "", "Command timed out", 1
def discover_cortex_skills():
"""Discover all available Cortex Code skills."""
print("Discovering Cortex Code capabilities...", file=sys.stderr)
# Run cortex skill list
stdout, stderr, code = run_command(["cortex", "skill", "list"])
if code != 0:
print(f"Error running cortex skill list: {stderr}", file=sys.stderr)
return {}
# Parse skill list output
skills = {}
# Handles two formats:
# Old format: "skill-name /path/to/skill"
# New format (v1.0.5.6+):
# [BUNDLED]
# - skill-name: /path/to/skill
for line in stdout.strip().split('\n'):
if not line.strip():
continue
# Skip section headers like [BUNDLED], [PROJECT], [GLOBAL]
if re.match(r'^\[.*\]$', line.strip()):
continue
# New format: " - skill-name: /path/to/skill"
new_format_match = re.match(r'^\s*-\s+(\S+?):\s+', line)
if new_format_match:
skill_name = new_format_match.group(1).strip()
else:
# Old format: "skill-name /path/to/skill"
parts = line.split()
if not parts:
continue
skill_name = parts[0].strip(':').strip()
# Read the skill's SKILL.md to get description and triggers
skill_info = read_skill_metadata(skill_name)
if skill_info:
skills[skill_name] = skill_info
return skills
def read_skill_metadata(skill_name):
"""Read SKILL.md frontmatter for a specific skill."""
# Cortex bundled skills are typically in ~/.local/share/cortex/{version}/bundled_skills/
cortex_share = Path.home() / ".local/share/cortex"
# Find the most recent version directory
if not cortex_share.exists():
return None
version_dirs = sorted([d for d in cortex_share.iterdir() if d.is_dir()], reverse=True)
for version_dir in version_dirs:
bundled_skills = version_dir / "bundled_skills"
if not bundled_skills.exists():
continue
# Look for skill directory
skill_path = bundled_skills / skill_name / "SKILL.md"
if skill_path.exists():
return parse_skill_md(skill_path)
return None
def parse_skill_md(skill_path):
"""Parse SKILL.md file and extract frontmatter."""
try:
with open(skill_path, 'r') as f:
content = f.read()
# Extract YAML frontmatter
frontmatter_match = re.match(r'^---\n(.*?)\n---', content, re.DOTALL)
if not frontmatter_match:
return None
frontmatter = frontmatter_match.group(1)
# Simple YAML parsing for name and description
name_match = re.search(r'name:\s*(.+)', frontmatter)
desc_match = re.search(r'description:\s*["\']?(.+?)["\']?$', frontmatter, re.MULTILINE | re.DOTALL)
if name_match and desc_match:
name = name_match.group(1).strip().strip('"\'')
description = desc_match.group(1).strip().strip('"\'')
# Extract "Use when" trigger patterns from body
triggers = extract_triggers(content)
return {
"name": name,
"description": description,
"triggers": triggers
}
except Exception as e:
print(f"Error parsing {skill_path}: {e}", file=sys.stderr)
return None
def extract_triggers(content):
"""Extract trigger phrases from skill content."""
triggers = []
# Look for "Use when", "Trigger", "When to use" sections
trigger_patterns = [
r'(?:Use when|When to use|Trigger).*?:\s*(.+?)(?=\n\n|\#\#)',
r'- Use (?:when|for|if):\s*(.+?)$'
]
for pattern in trigger_patterns:
matches = re.finditer(pattern, content, re.MULTILINE | re.DOTALL)
for match in matches:
trigger_text = match.group(1).strip()
# Clean up and split by common separators
phrases = re.split(r'[,;]|\n-', trigger_text)
triggers.extend([p.strip() for p in phrases if p.strip()])
return triggers[:10] # Limit to 10 most relevant triggers
def main():
"""Main discovery function."""
# Parse command line arguments
parser = argparse.ArgumentParser(description="Discover Cortex Code capabilities")
parser.add_argument(
"--cache-dir",
type=Path,
help="Cache directory for storing capabilities (default: from config or ~/.cache/cortex-skill)"
)
args = parser.parse_args()
# Determine cache directory
if args.cache_dir:
cache_dir = args.cache_dir
else:
# Get default from config
config_manager = ConfigManager()
cache_dir_str = config_manager.get("security.cache_dir")
cache_dir = Path(cache_dir_str).expanduser()
# Discover capabilities
capabilities = discover_cortex_skills()
# Cache using CacheManager with SHA256 fingerprint validation
try:
cache_manager = CacheManager(cache_dir)
cache_manager.write("cortex-capabilities", capabilities, ttl=86400) # 24-hour TTL
print(f"Discovered {len(capabilities)} Cortex skills", file=sys.stderr)
print(f"Cached to: {cache_dir / 'cortex-capabilities.json'}", file=sys.stderr)
except Exception as e:
# If cache fails, log warning but continue
print(f"Warning: Failed to cache capabilities: {e}", file=sys.stderr)
print(f"Discovered {len(capabilities)} Cortex skills", file=sys.stderr)
# Output the capabilities
print(json.dumps(capabilities, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/execute_cortex.py
#!/usr/bin/env python3
"""
Executes Cortex Code in headless mode with streaming output parsing.
Uses --output-format stream-json for streaming results.
Handles tool use events and final results.
"""
import json
import os
import subprocess
import sys
import argparse
import threading
import queue
import time
from pathlib import Path
from typing import List, Dict, Optional
try:
from security.prompt_sanitizer import PromptSanitizer
except Exception:
PromptSanitizer = None
# Known tools for inversion logic (allowed -> disallowed)
KNOWN_TOOLS = [
"Read", "Write", "Edit", "Bash", "Grep", "Glob",
"snowflake_sql_execute", "data_diff", "snowflake_query"
]
DESTRUCTIVE_SHELL_TOOLS = [
"Bash",
"Bash(rm *)", "Bash(rm -rf *)", "Bash(rm -r *)",
"Bash(sudo *)", "Bash(chmod 777 *)",
"Bash(git push *)", "Bash(git reset --hard *)"
]
READ_ONLY_TOOLS = ["Edit", "Write", "Bash"] + DESTRUCTIVE_SHELL_TOOLS
UNKNOWN_TOOL_SENTINEL = "*"
def _redact_error_output(error_text: str) -> str:
"""Redact sensitive data before returning/logging error output."""
if PromptSanitizer is None:
return error_text
return PromptSanitizer().sanitize(error_text)
def invert_tools_to_disallowed(allowed_tools: List[str]) -> List[str]:
"""
Convert allowed tools list to disallowed tools list.
For prompt mode: when security wrapper predicts/approves specific tools,
we need to invert the list to block all OTHER tools via --disallowed-tools.
Args:
allowed_tools: List of tool names that ARE allowed
Returns:
List of tool names that should be disallowed (inverse of allowed)
Example:
allowed = ["Read", "Grep"]
disallowed = ["Write", "Edit", "Bash", "Glob", ...other tools...]
"""
inverted = [tool for tool in KNOWN_TOOLS if tool not in allowed_tools]
inverted.append(UNKNOWN_TOOL_SENTINEL)
return inverted
def execute_cortex_streaming(prompt: str, connection: Optional[str] = None,
disallowed_tools: Optional[List[str]] = None,
envelope: str = "RW",
approval_mode: str = "prompt",
allowed_tools: Optional[List[str]] = None,
timeout_seconds: int = 300,
deploy_confirmed: bool = False) -> Dict:
"""
Execute Cortex with streaming JSON output in programmatic mode.
Uses --output-format stream-json for streaming results.
Tools are controlled via --disallowed-tools blocklists for safety.
Args:
prompt: The enriched prompt to send to Cortex
connection: Optional Snowflake connection name
disallowed_tools: Optional list of tools to explicitly block
envelope: Security envelope mode (RO, RW, RESEARCH, DEPLOY, NONE)
approval_mode: Approval mode (prompt, auto, envelope_only)
allowed_tools: Optional list of tools that ARE allowed (for prompt mode)
Returns:
Dictionary with execution results
"""
if approval_mode in ["auto", "envelope_only"] and envelope == "NONE":
raise ValueError("NONE envelope is not allowed in auto or envelope_only approval modes")
if approval_mode in ["auto", "envelope_only"] and envelope == "DEPLOY" and not deploy_confirmed:
raise ValueError("DEPLOY envelope requires explicit confirmation")
# Build command in print mode. The prompt is delivered with -p; do not add
# --input-format stream-json here. Cortex treats that flag as JSON stdin
# input mode, so combining it with -p and closed stdin can emit only the
# initial session event and exit before the prompt is processed.
cmd = [
"cortex",
"-p", prompt,
"--output-format", "stream-json"
]
# Add connection if specified
if connection:
cmd.extend(["-c", connection])
# Step 1: Handle approval mode — build disallowed tools list for envelope security.
# Do NOT use --allowed-tools: it creates a "must match pattern" check that
# blocks Snowflake MCP tools.
final_disallowed_tools = disallowed_tools or []
if approval_mode == "prompt":
# Prompt mode: invert allowed_tools to disallowed_tools
# In prompt mode, we ONLY use allowed_tools (don't merge with envelope)
if allowed_tools is not None:
# User approved specific tools - block everything else
inverted_tools = invert_tools_to_disallowed(allowed_tools)
# Merge with existing disallowed tools (but NOT envelope tools)
final_disallowed_tools = list(set(final_disallowed_tools) | set(inverted_tools))
else:
# No tools approved - block all known tools
final_disallowed_tools = list(set(final_disallowed_tools) | set(KNOWN_TOOLS))
elif approval_mode in ["envelope_only", "auto"]:
# Envelope-only or auto mode: apply envelope-based security via blocklist.
envelope_tools = []
if envelope == "RO":
# Read-only: block all write operations
envelope_tools = READ_ONLY_TOOLS
elif envelope in ["RW", "DEPLOY"]:
# RW and DEPLOY may allow shell usage, but still block destructive
# shell patterns by default. Explicit custom disallowed_tools can
# add stricter policy on top.
envelope_tools = DESTRUCTIVE_SHELL_TOOLS
elif envelope == "RESEARCH":
# Research: read-only plus web access
envelope_tools = READ_ONLY_TOOLS
# Merge envelope tools with final disallowed list
if envelope_tools:
final_disallowed_tools = list(set(final_disallowed_tools) | set(envelope_tools))
# Step 3: Add final disallowed tools to command
if final_disallowed_tools:
for tool in final_disallowed_tools:
cmd.extend(["--disallowed-tools", tool])
debug_cmd = f"cortex -p \"...\" --output-format stream-json"
if connection:
debug_cmd += f" -c {connection}"
if final_disallowed_tools:
debug_cmd += f" --disallowed-tools {' '.join(final_disallowed_tools[:3])}{'...' if len(final_disallowed_tools) > 3 else ''}"
print(debug_cmd, file=sys.stderr)
process = None
stderr_lines = []
def _read_stderr(stderr):
if stderr is None:
return
for stderr_line in stderr:
stderr_lines.append(stderr_line)
def _kill_process():
if not process:
return
process.kill()
try:
process.wait(timeout=1)
except Exception:
pass
try:
# Start process. stdin=DEVNULL prevents accidental reads from the parent
# terminal; prompt delivery is handled exclusively by -p print mode.
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
stdin=subprocess.DEVNULL,
text=True,
bufsize=1
)
stderr_thread = threading.Thread(target=_read_stderr, args=(process.stderr,), daemon=True)
stderr_thread.start()
results = {
"session_id": None,
"events": [],
"permission_requests": [],
"final_result": None,
"error": None
}
stdout_queue = queue.Queue()
stdout_errors = queue.Queue()
def _read_stdout(stdout):
if stdout is None:
stdout_queue.put(None)
return
try:
for stdout_line in stdout:
stdout_queue.put(stdout_line)
except Exception as exc:
stdout_errors.put(exc)
finally:
stdout_queue.put(None)
stdout_thread = threading.Thread(target=_read_stdout, args=(process.stdout,), daemon=True)
stdout_thread.start()
timed_out = False
deadline = time.monotonic() + timeout_seconds
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
timed_out = True
break
try:
line = stdout_queue.get(timeout=remaining)
except queue.Empty:
timed_out = True
break
if line is None:
if not stdout_errors.empty():
raise stdout_errors.get()
break
if not line.strip():
continue
try:
event = json.loads(line)
results["events"].append(event)
event_type = event.get("type")
# Extract session ID
if event_type == "system" and event.get("subtype") == "init":
results["session_id"] = event.get("session_id")
print(f"→ Started Cortex session: {results['session_id']}", file=sys.stderr)
# Handle assistant responses
elif event_type == "assistant":
message = event.get("message", {})
content = message.get("content", [])
for item in content:
if item.get("type") == "text":
print(f"[Cortex] {item.get('text', '')}", file=sys.stderr)
elif item.get("type") == "tool_use":
tool_name = item.get("name")
print(f"[Cortex] Using tool: {tool_name}", file=sys.stderr)
# Handle permission requests (via user messages with tool_result containing denials)
elif event_type == "user":
message = event.get("message", {})
content = message.get("content", [])
for item in content:
if item.get("type") == "tool_result":
tool_content = item.get("content", "")
tool_content_text = json.dumps(tool_content) if isinstance(tool_content, list) else str(tool_content)
if "Permission denied" in tool_content_text or "denied" in tool_content_text.lower():
results["permission_requests"].append({
"tool_use_id": item.get("tool_use_id"),
"content": tool_content
})
print(f"[Cortex] Permission request detected: {tool_content_text}", file=sys.stderr)
# Handle final result
elif event_type == "result":
results["final_result"] = event.get("result")
print(f"[Cortex] Result: {event.get('result')}", file=sys.stderr)
except json.JSONDecodeError as e:
print(f"Warning: Failed to parse line: {line[:100]}... Error: {e}", file=sys.stderr)
continue
if timed_out:
raise subprocess.TimeoutExpired(cmd=cmd, timeout=timeout_seconds)
# Wait for process to complete
process.wait(timeout=timeout_seconds)
stderr_thread.join(timeout=1)
# Check for errors
if process.returncode != 0:
stderr_output = _redact_error_output("".join(stderr_lines))
results["error"] = stderr_output
print(f"Error: Cortex exited with code {process.returncode}", file=sys.stderr)
print(f"Stderr: {stderr_output}", file=sys.stderr)
return results
except subprocess.TimeoutExpired:
_kill_process()
return {
"session_id": None,
"events": [],
"permission_requests": [],
"final_result": None,
"error": f"Cortex execution timed out after {timeout_seconds} seconds"
}
except Exception as e:
_kill_process()
return {
"session_id": None,
"events": [],
"permission_requests": [],
"final_result": None,
"error": _redact_error_output(str(e))
}
def _resolve_output_path(output_file: str) -> Path:
"""Resolve output path under a safe output directory."""
base_dir = Path(os.environ.get("CORTEX_CODE_OUTPUT_DIR", Path.cwd())).expanduser().resolve()
output_path = Path(output_file).expanduser()
if not output_path.is_absolute():
output_path = base_dir / output_path
output_path = output_path.resolve()
try:
output_path.relative_to(base_dir)
except ValueError as exc:
raise ValueError(f"Output file must be under {base_dir}") from exc
return output_path
def main():
"""Main execution function."""
parser = argparse.ArgumentParser(description="Execute Cortex Code headlessly")
parser.add_argument("--prompt", required=True, help="Prompt to send to Cortex")
parser.add_argument("--connection", "-c", help="Snowflake connection name")
parser.add_argument("--disallowed-tools", nargs="+", help="Tools to explicitly block")
parser.add_argument("--envelope", default="RW",
choices=["RO", "RW", "RESEARCH", "DEPLOY", "NONE"],
help="Security envelope mode (default: RW)")
parser.add_argument("--approval-mode", default="prompt",
choices=["prompt", "auto", "envelope_only"],
help="Approval mode (default: prompt)")
parser.add_argument("--allowed-tools", nargs="+",
help="Tools that are allowed (for prompt mode)")
parser.add_argument("--timeout", type=int, default=300,
help="Maximum seconds to wait for Cortex execution (default: 300)")
parser.add_argument("--deploy-confirmed", action="store_true",
help="Required explicit confirmation for DEPLOY envelope in non-interactive modes")
parser.add_argument("--output-file", help="Write JSON results to this file instead of stdout")
parser.add_argument("--stream", action="store_true", help="Stream output (always true)")
args = parser.parse_args()
# Execute Cortex
results = execute_cortex_streaming(
args.prompt,
connection=args.connection,
disallowed_tools=args.disallowed_tools,
envelope=args.envelope,
approval_mode=args.approval_mode,
allowed_tools=args.allowed_tools,
timeout_seconds=args.timeout,
deploy_confirmed=args.deploy_confirmed
)
# Output results as JSON
output = json.dumps(results, indent=2)
if args.output_file:
try:
output_path = _resolve_output_path(args.output_file)
except ValueError as exc:
print(json.dumps({"error": str(exc)}, indent=2))
return 1
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(output + "\n")
else:
print(output)
# Exit with appropriate code
if results.get("error"):
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/predict_tools.py
#!/usr/bin/env python3
"""
Predicts which Cortex tools will be needed based on the user prompt and capabilities.
Enhanced with confidence scoring for approval handler.
"""
import json
import sys
import argparse
from pathlib import Path
from security.cache_manager import CacheManager
from security.config_manager import ConfigManager
# Tool prediction mappings with weighted patterns
TOOL_PATTERNS = {
"snowflake_sql_execute": [
"select", "insert", "update", "delete", "query", "sql",
"table", "database", "data", "snowflake"
],
"bash": [
"run", "execute", "command", "script", "install", "shell"
],
"read": [
"read", "show", "display", "view", "check", "inspect", "examine"
],
"write": [
"create", "write", "generate", "save", "output", "file"
],
"glob": [
"find", "search", "list", "files", "directory", "locate"
],
"grep": [
"search", "find", "pattern", "match", "contains"
]
}
# Always include these base tools for Snowflake operations
BASE_SNOWFLAKE_TOOLS = ["snowflake_sql_execute", "bash", "read"]
def load_capabilities():
"""Load cached Cortex capabilities through CacheManager."""
try:
config_manager = ConfigManager()
cache_dir = Path(config_manager.get("security.cache_dir")).expanduser()
cache_manager = CacheManager(cache_dir)
return cache_manager.read("cortex-capabilities") or {}
except Exception as exc:
print(f"Warning: Failed to load Cortex capabilities from cache: {exc}", file=sys.stderr)
return {}
def predict_tools(prompt, envelope=None):
"""
Predict required tools based on prompt analysis with confidence scoring.
Args:
prompt: User prompt to analyze
envelope: Optional envelope dict with capabilities
Returns:
dict with:
- tools: list of predicted tool names
- confidence: float 0-1 indicating prediction confidence
- reasoning: str explaining the prediction
"""
prompt_lower = prompt.lower()
predicted = set(BASE_SNOWFLAKE_TOOLS)
matched_patterns = []
# Check each tool pattern and track matches
for tool, patterns in TOOL_PATTERNS.items():
tool_matches = []
for pattern in patterns:
if pattern in prompt_lower:
tool_matches.append(pattern)
if tool_matches:
predicted.add(tool)
matched_patterns.append(f"{tool}: {', '.join(tool_matches)}")
# Calculate confidence based on pattern matches
total_words = len(prompt_lower.split())
pattern_match_count = len(matched_patterns)
# Base confidence on match density
if total_words == 0:
confidence = 0.5
elif pattern_match_count == 0:
# Only base tools predicted
confidence = 0.5
else:
# More matches relative to prompt length = higher confidence
confidence = min(0.9, 0.5 + (pattern_match_count / max(total_words / 5, 1)) * 0.4)
# Adjust confidence based on prompt clarity
if total_words < 5:
confidence *= 0.8 # Short prompts are less clear
elif total_words > 20:
confidence *= 0.95 # Very detailed prompts slightly less confident
# Check capabilities if provided in envelope
if envelope and "capabilities" in envelope:
capabilities = envelope["capabilities"]
for skill_name, skill_info in capabilities.items():
description = skill_info.get("description", "").lower()
# If skill description matches prompt, boost confidence
if any(word in description for word in prompt_lower.split()):
confidence = min(1.0, confidence + 0.1)
# Data quality skills typically need more tools
if "quality" in skill_name or "governance" in skill_name:
predicted.update(["glob", "grep", "write"])
matched_patterns.append(f"skill_match: {skill_name}")
# ML skills need bash for model operations
if "ml" in skill_name or "machine" in skill_name or "forecast" in skill_name:
predicted.add("bash")
matched_patterns.append(f"skill_match: {skill_name}")
# Generate reasoning
if matched_patterns:
reasoning = f"Matched {len(matched_patterns)} patterns: {'; '.join(matched_patterns[:3])}"
if len(matched_patterns) > 3:
reasoning += f" and {len(matched_patterns) - 3} more"
else:
reasoning = "Using base Snowflake tools only - no specific patterns matched"
return {
"tools": sorted(list(predicted)),
"confidence": round(confidence, 2),
"reasoning": reasoning
}
def main():
"""Main tool prediction function."""
parser = argparse.ArgumentParser(description="Predict required Cortex tools")
parser.add_argument("--prompt", required=True, help="User prompt to analyze")
args = parser.parse_args()
# Load capabilities
capabilities = load_capabilities()
envelope = {"capabilities": capabilities} if capabilities else None
# Predict tools with confidence
result = predict_tools(args.prompt, envelope)
# Output as JSON
print(json.dumps(result, indent=2))
# Summary to stderr
print(f"\nPredicted {len(result['tools'])} tools with {result['confidence']:.0%} confidence:", file=sys.stderr)
print(f" Tools: {', '.join(result['tools'])}", file=sys.stderr)
print(f" Reasoning: {result['reasoning']}", file=sys.stderr)
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/read_cortex_sessions.py
#!/usr/bin/env python3
"""
Reads recent Cortex Code session files for context enrichment.
"""
import json
import sys
import argparse
from pathlib import Path
from datetime import datetime
MAX_SESSION_BYTES = 5 * 1024 * 1024
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from security.prompt_sanitizer import PromptSanitizer
def find_recent_sessions(limit=3):
"""Find the most recent Cortex session files."""
sessions_dir = Path.home() / ".local/share/cortex/sessions"
if not sessions_dir.exists():
print(f"Sessions directory not found: {sessions_dir}", file=sys.stderr)
return []
# Find all .jsonl session files
session_files = sorted(
[f for f in sessions_dir.glob("**/*.jsonl")],
key=lambda f: f.stat().st_mtime,
reverse=True
)
return session_files[:limit]
def parse_session_file(session_path, sanitize=True):
"""Parse a session JSONL file and extract key information.
Args:
session_path: Path to the session JSONL file
sanitize: Whether to sanitize PII from text content (default: True)
Returns:
Dictionary with session data, or None on error
"""
try:
if session_path.stat().st_size > MAX_SESSION_BYTES:
print(f"Skipping oversized session file: {session_path}", file=sys.stderr)
return None
# Initialize sanitizer if needed
sanitizer = PromptSanitizer() if sanitize else None
session_data = {
"session_id": None,
"timestamp": session_path.stat().st_mtime,
"user_prompts": [],
"assistant_responses": [],
"tools_used": [],
"result": None
}
with open(session_path, 'r') as f:
for line in f:
if not line.strip():
continue
try:
event = json.loads(line)
event_type = event.get("type")
if event_type == "system" and event.get("subtype") == "init":
session_data["session_id"] = event.get("session_id")
elif event_type == "user":
# Check if this is a tool result or user message
message = event.get("message", {})
content = message.get("content", [])
# Extract user text if present
for item in content:
if item.get("type") == "text":
text = item.get("text", "")
# Sanitize user prompts if enabled
if sanitizer:
text = sanitizer.sanitize(text)
session_data["user_prompts"].append(text)
elif event_type == "assistant":
message = event.get("message", {})
content = message.get("content", [])
for item in content:
if item.get("type") == "text":
text = item.get("text", "")
# Sanitize assistant responses if enabled
if sanitizer:
text = sanitizer.sanitize(text)
session_data["assistant_responses"].append(text)
elif item.get("type") == "tool_use":
tool_name = item.get("name")
if tool_name:
session_data["tools_used"].append(tool_name)
elif event_type == "result":
session_data["result"] = event.get("result")
except json.JSONDecodeError:
continue
return session_data
except Exception as e:
print(f"Error parsing session {session_path}: {e}", file=sys.stderr)
return None
def summarize_sessions(session_files, sanitize=True):
"""Summarize recent Cortex sessions.
Args:
session_files: List of session file paths
sanitize: Whether to sanitize PII from text content (default: True)
Returns:
List of session summary dictionaries
"""
summaries = []
for session_path in session_files:
session_data = parse_session_file(session_path, sanitize=sanitize)
if not session_data:
continue
# Create a concise summary
# Note: session_data already has sanitized content if sanitize=True
summary = {
"file": session_path.name,
"session_id": session_data["session_id"],
"time": datetime.fromtimestamp(session_data["timestamp"]).strftime("%Y-%m-%d %H:%M:%S"),
"prompts_count": len(session_data["user_prompts"]),
"tools_used": list(set(session_data["tools_used"])),
"last_prompt": session_data["user_prompts"][-1] if session_data["user_prompts"] else None,
"result_type": type(session_data["result"]).__name__ if session_data["result"] else None
}
summaries.append(summary)
return summaries
def main():
"""Main function to read and summarize recent Cortex sessions."""
parser = argparse.ArgumentParser(description="Read recent Cortex sessions")
parser.add_argument("--limit", type=int, default=3, help="Number of recent sessions to read")
parser.add_argument("--verbose", action="store_true", help="Include full session details")
parser.add_argument("--no-sanitize", action="store_true", help="Disable PII sanitization (for debugging)")
args = parser.parse_args()
# Determine if sanitization should be enabled (default: True)
sanitize = not args.no_sanitize
# Find recent sessions
session_files = find_recent_sessions(args.limit)
if not session_files:
print("No recent Cortex sessions found", file=sys.stderr)
return 0
print(f"Found {len(session_files)} recent sessions", file=sys.stderr)
# Summarize sessions with sanitization flag
summaries = summarize_sessions(session_files, sanitize=sanitize)
# Output JSON
print(json.dumps(summaries, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/route_request.py
#!/usr/bin/env python3
"""
LLM-based routing logic to determine if request should go to Cortex Code or Codex.
Uses semantic understanding rather than simple keyword matching.
"""
import json
import sys
import argparse
import fnmatch
import re
from pathlib import Path
from typing import Optional, Dict, Any
# Add parent directory to path for security imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from security.config_manager import ConfigManager
from security.cache_manager import CacheManager
# Snowflake/Cortex indicators
SNOWFLAKE_INDICATORS = [
"snowflake", "cortex", "warehouse", "snowpark", "data warehouse",
"cortex ai", "cortex search", "cortex analyst", "dynamic table",
"snowflake database", "snowflake schema", "snowflake table",
"data governance", "data quality", "trust my data",
"ml function", "classification", "forecasting"
]
# Non-Snowflake indicators (route to Codex)
SNOWFLAKE_CONTEXT_TERMS = ["snowflake", "warehouse", "cortex", "schema", "table", "database"]
AMBIGUOUS_SNOWFLAKE_TERMS = ["stream", "task", "stage", "pipe"]
PATH_TOKEN_PATTERN = re.compile(r'(?<![\w.-])(?:~/?|/|\./|\.\./|[A-Za-z0-9_.-]+/)[A-Za-z0-9_./$~:-]+|(?<![\w.-])(?:\.ssh|\.aws|\.snowflake|\.env(?:\.[\w-]+)?|credentials\.(?:json|ya?ml)|[A-Za-z0-9_.-]+_key\.(?:p8|pem))(?![\w.-])', re.IGNORECASE)
CLAUDE_CODE_INDICATORS = [
"local file", "git", "github", "commit", "push", "pull request",
"python script", "javascript", "react", "frontend", "backend",
"postgres", "mysql", "mongodb", "redis",
"docker", "kubernetes", "infrastructure",
"read file", "write file", "edit file", "create file"
]
# Backwards-compatible name used by shared tests and copied integrations.
CODING_AGENT_INDICATORS = CLAUDE_CODE_INDICATORS
def load_cortex_capabilities():
"""Load cached Cortex capabilities using CacheManager."""
try:
# Get cache directory from config
config_manager = ConfigManager()
cache_dir_str = config_manager.get("security.cache_dir")
cache_dir = Path(cache_dir_str).expanduser()
# Use CacheManager to read cache with integrity validation
cache_manager = CacheManager(cache_dir)
capabilities = cache_manager.read("cortex-capabilities")
if capabilities is None:
print("Warning: Cortex capabilities not cached. Run discover_cortex.py first.", file=sys.stderr)
return {}
return capabilities
except Exception as e:
print(f"Warning: Failed to load Cortex capabilities from cache: {e}", file=sys.stderr)
print("Run discover_cortex.py to cache capabilities.", file=sys.stderr)
return {}
def analyze_with_llm_logic(prompt, capabilities):
"""
Analyze prompt using LLM-inspired logic.
This is a deterministic approximation of what an LLM would consider.
"""
prompt_lower = prompt.lower()
# Score based on indicators
snowflake_score = 0
claude_score = 0
# Check for explicit Snowflake/Cortex mentions
for indicator in SNOWFLAKE_INDICATORS:
if indicator in prompt_lower:
snowflake_score += 3 if indicator in ["snowflake", "cortex"] else 1
# Ambiguous Snowflake object names only count with Snowflake context.
if any(context in prompt_lower for context in SNOWFLAKE_CONTEXT_TERMS):
for term in AMBIGUOUS_SNOWFLAKE_TERMS:
if term in prompt_lower:
snowflake_score += 1
# Check for non-Snowflake indicators
for indicator in CLAUDE_CODE_INDICATORS:
if indicator in prompt_lower:
claude_score += 2
# Check against Cortex skill triggers
for skill_name, skill_info in capabilities.items():
for trigger in skill_info.get("triggers", []):
trigger_lower = trigger.lower()
if trigger_lower in prompt_lower or any(word in prompt_lower for word in trigger_lower.split()):
snowflake_score += 2
break
# SQL query detection
sql_keywords = ["select", "insert", "update", "delete", "create table", "alter", "drop"]
if any(kw in prompt_lower for kw in sql_keywords):
# Could be any database, but check for Snowflake context
if any(ind in prompt_lower for ind in ["snowflake", "warehouse", "cortex"]):
snowflake_score += 3
else:
# Generic SQL, likely not Snowflake
claude_score += 1
# Data-related terms (ambiguous, need context)
data_terms = ["data quality", "schema", "table", "database", "query"]
data_term_count = sum(1 for term in data_terms if term in prompt_lower)
if data_term_count >= 2:
# Multiple data terms suggest database work
# Check if Snowflake context exists
if snowflake_score > 0:
snowflake_score += 2
# Calculate confidence
total_score = snowflake_score + claude_score
if total_score == 0:
# No strong indicators, default to the host coding agent for safety.
# Install scripts replace this placeholder with claude/codex/cursor.
return "__CODING_AGENT__", 0.5
confidence = max(snowflake_score, claude_score) / total_score
if snowflake_score > claude_score:
return "cortex", confidence
else:
return "__CODING_AGENT__", confidence
def check_credential_allowlist(
prompt: str,
config_path: Optional[Path] = None,
org_policy_path: Optional[Path] = None
) -> Dict[str, Any]:
"""
Check if prompt contains credential file paths from the allowlist.
This function runs before routing analysis to block prompts that reference
credential files, regardless of whether they would be routed to Cortex or Codex.
Args:
prompt: User prompt to check
config_path: Path to user config file (optional)
org_policy_path: Path to organization policy file (optional)
Returns:
Dict with blocking decision:
- blocked: True if credential detected, False otherwise
- route: "blocked" if blocked, None otherwise
- confidence: 1.0 if blocked (100% confident in blocking)
- reason: Human-readable reason for blocking
- pattern_matched: The allowlist pattern that matched
"""
# Initialize ConfigManager with optional config paths
config_manager = ConfigManager(
config_path=config_path,
org_policy_path=org_policy_path
)
# Load credential allowlist
credential_allowlist = config_manager.get("security.credential_file_allowlist")
prompt_tokens = PATH_TOKEN_PATTERN.findall(prompt)
normalized_tokens = []
for token in prompt_tokens:
normalized_tokens.append(token)
if token.startswith("~"):
normalized_tokens.append(token.replace("~", str(Path.home()), 1))
for pattern in credential_allowlist:
expanded_pattern = str(Path(pattern).expanduser())
candidate_patterns = [pattern, expanded_pattern]
if pattern.startswith("~/**/"):
candidate_patterns.append("**/" + pattern.split("~/**/", 1)[1])
for token in normalized_tokens:
token_lower = token.lower()
for candidate_pattern in candidate_patterns:
pattern_lower = candidate_pattern.lower()
pattern_dir = pattern_lower.split("*")[0].rstrip("/")
if (
fnmatch.fnmatch(token_lower, pattern_lower)
or fnmatch.fnmatch(f"*/{token_lower}", pattern_lower)
or (token_lower in {".ssh", ".aws", ".snowflake"} and pattern_dir.endswith(token_lower))
):
return {
"blocked": True,
"route": "blocked",
"confidence": 1.0,
"reason": f"Prompt contains credential file path from allowlist",
"pattern_matched": pattern
}
# No credentials detected
return {
"blocked": False
}
def main():
"""Main routing function."""
parser = argparse.ArgumentParser(description="Route request to Cortex or Codex")
parser.add_argument("--prompt", required=True, help="User prompt to analyze")
parser.add_argument("--config", help="Path to user config file")
parser.add_argument("--org-policy", help="Path to organization policy file")
args = parser.parse_args()
# Step 1: Check credential allowlist BEFORE routing
config_path = Path(args.config) if args.config else None
org_policy_path = Path(args.org_policy) if args.org_policy else None
credential_check = check_credential_allowlist(
args.prompt,
config_path,
org_policy_path
)
# If blocked by credential check, return immediately
if credential_check.get("blocked"):
print(json.dumps(credential_check, indent=2))
print(f"\n⛔ BLOCKED: Credential file detected", file=sys.stderr)
print(f" Pattern: {credential_check['pattern_matched']}", file=sys.stderr)
print(f" Reason: {credential_check['reason']}", file=sys.stderr)
sys.exit(0)
# Step 2: Load Cortex capabilities
capabilities = load_cortex_capabilities()
# Step 3: Analyze prompt for routing
route, confidence = analyze_with_llm_logic(args.prompt, capabilities)
# Step 4: Output decision
result = {
"route": route,
"confidence": confidence,
"reasoning": f"Routed to {route} with {confidence:.2%} confidence"
}
print(json.dumps(result, indent=2))
print(f"\n→ Route to: {route.upper()}", file=sys.stderr)
print(f" Confidence: {confidence:.2%}", file=sys.stderr)
sys.exit(0)
if __name__ == "__main__":
main()
scripts/security_wrapper.py
#!/usr/bin/env python3
"""
Security wrapper orchestrator for cortex-code skill.
Coordinates all security components:
- ConfigManager: Load and validate configuration
- AuditLogger: Log all executions
- CacheManager: Secure caching
- PromptSanitizer: Remove PII and detect injection
- ApprovalHandler: Tool prediction and user approval
This is the main entry point for secure Cortex execution.
"""
import argparse
import fnmatch
import json
import re
import sys
import os
from pathlib import Path
from typing import Optional, Dict, Any
# Add parent directories to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from security.config_manager import ConfigManager
from security.audit_logger import AuditLogger
from security.cache_manager import CacheManager
from security.prompt_sanitizer import PromptSanitizer
from security.approval_handler import ApprovalHandler
# Import routing functions
sys.path.insert(0, str(Path(__file__).parent))
from route_request import analyze_with_llm_logic, load_cortex_capabilities
from execute_cortex import execute_cortex_streaming
def _log_audit_event(audit_logger, **kwargs):
"""Best-effort audit logging helper."""
try:
return audit_logger.log_execution(**kwargs), None
except Exception as exc:
print(f"Warning: failed to write audit log: {exc}", file=sys.stderr)
return None, str(exc)
PATH_TOKEN_PATTERN = re.compile(r'(?<![\w.-])(?:~/?|/|\./|\.\./|[A-Za-z0-9_.-]+/)[A-Za-z0-9_./$~:-]+|(?<![\w.-])(?:\.ssh|\.aws|\.snowflake|\.env(?:\.[\w-]+)?|credentials\.(?:json|ya?ml)|[A-Za-z0-9_.-]+_key\.(?:p8|pem))(?![\w.-])', re.IGNORECASE)
def execute_with_security(
prompt: str,
config_path: Optional[str] = None,
org_policy_path: Optional[str] = None,
dry_run: bool = False,
envelope: Optional[Dict[str, Any]] = None,
mock_user_approval: Optional[str] = None
) -> Dict[str, Any]:
"""
Execute prompt with full security orchestration.
This function:
1. Loads configuration (with org policy override)
2. Initializes all security components
3. Sanitizes prompt if enabled
4. Determines approval mode
5. In dry-run mode: returns initialization status
6. In live mode: Full execution with approval flow
Args:
prompt: User prompt to execute
config_path: Path to user config file (optional)
org_policy_path: Path to organization policy file (optional)
dry_run: If True, only initialize and validate (don't execute)
envelope: Cortex envelope dict (optional)
mock_user_approval: For testing - "approve" or "deny" (optional)
Returns:
Dict with execution results or initialization status
"""
# Step 1: Load configuration
config_path_obj = Path(config_path) if config_path else None
org_policy_path_obj = Path(org_policy_path) if org_policy_path else None
config_manager = ConfigManager(
config_path=config_path_obj,
org_policy_path=org_policy_path_obj
)
# Extract config values
approval_mode = config_manager.get("security.approval_mode")
audit_log_path = Path(config_manager.get("security.audit_log_path"))
audit_log_rotation = config_manager.get("security.audit_log_rotation")
audit_log_retention = config_manager.get("security.audit_log_retention")
cache_dir = Path(config_manager.get("security.cache_dir"))
sanitize_enabled = config_manager.get("security.sanitize_conversation_history")
confidence_threshold = config_manager.get("security.tool_prediction_confidence_threshold")
allowed_envelopes = config_manager.get("security.allowed_envelopes")
# Step 2: Initialize security components
audit_logger = AuditLogger(
log_path=audit_log_path,
rotation_size=audit_log_rotation,
retention_days=audit_log_retention
)
cache_manager = CacheManager(cache_dir=cache_dir)
prompt_sanitizer = PromptSanitizer()
approval_handler = ApprovalHandler(confidence_threshold=confidence_threshold)
# Step 3: Sanitize prompt if enabled
sanitized_prompt = prompt
if sanitize_enabled:
sanitized_prompt = prompt_sanitizer.sanitize(prompt)
if sanitized_prompt == "[POTENTIAL INJECTION DETECTED - REMOVED]":
return {
"status": "blocked",
"reason": "Prompt injection attempt detected",
"message": "Cannot route prompts containing prompt injection attempts",
"sanitized_prompt": sanitized_prompt
}
envelope_mode = "RW"
if isinstance(envelope, dict):
envelope_mode = envelope.get("mode") or envelope.get("type") or "RW"
elif isinstance(envelope, str):
envelope_mode = envelope
if envelope_mode not in allowed_envelopes:
return {
"status": "blocked",
"reason": f"Envelope {envelope_mode} is not allowed by configuration",
"allowed_envelopes": allowed_envelopes,
"requested_envelope": envelope_mode,
}
# Step 4: Check credential file allowlist (on original prompt)
credential_allowlist = config_manager.get("security.credential_file_allowlist")
prompt_tokens = PATH_TOKEN_PATTERN.findall(prompt)
normalized_tokens = []
for token in prompt_tokens:
normalized_tokens.append(token)
if token.startswith("~"):
normalized_tokens.append(token.replace("~", str(Path.home()), 1))
for pattern in credential_allowlist:
expanded_pattern = str(Path(pattern).expanduser())
candidate_patterns = [pattern, expanded_pattern]
if pattern.startswith("~/**/"):
candidate_patterns.append("**/" + pattern.split("~/**/", 1)[1])
for token in normalized_tokens:
token_lower = token.lower()
for candidate_pattern in candidate_patterns:
pattern_lower = candidate_pattern.lower()
pattern_dir = pattern_lower.split("*")[0].rstrip("/")
if (
fnmatch.fnmatch(token_lower, pattern_lower)
or fnmatch.fnmatch(f"*/{token_lower}", pattern_lower)
or (token_lower in {".ssh", ".aws", ".snowflake"} and pattern_dir.endswith(token_lower))
):
return {
"status": "blocked",
"reason": "Prompt contains credential file path from allowlist",
"pattern_matched": pattern,
"message": "Cannot route prompts containing credential file paths for security"
}
# Step 5: Determine routing (cortex vs claude) on sanitized prompt
capabilities = load_cortex_capabilities()
route_decision, route_confidence = analyze_with_llm_logic(sanitized_prompt, capabilities)
# Step 6: Determine approval mode
# In prompt mode, user must approve tools
# In auto mode, tools are auto-approved
# In deny mode, execution is blocked
# Step 7: Dry-run mode - return initialization status
if dry_run:
return {
"status": "initialized",
"dry_run": True,
"sanitized_prompt": sanitized_prompt,
"routing": {
"decision": route_decision,
"confidence": route_confidence
},
"config": {
"approval_mode": approval_mode,
"audit_log_path": str(audit_log_path),
"cache_dir": str(cache_dir),
"sanitize_enabled": sanitize_enabled,
"confidence_threshold": confidence_threshold,
"allowed_envelopes": allowed_envelopes
},
"audit_logger": str(type(audit_logger).__name__),
"cache_manager": str(type(cache_manager).__name__),
"prompt_sanitizer": str(type(prompt_sanitizer).__name__),
"approval_handler": str(type(approval_handler).__name__)
}
# Step 8: Full execution flow
# Route to Coding Agent for non-Snowflake requests
if route_decision == "__CODING_AGENT__":
return {
"status": "routed_to_coding_agent",
"message": "Request routed to coding agent for local handling",
"routing": {"decision": route_decision, "confidence": route_confidence}
}
# Step 9: Tool prediction for Cortex execution
prediction = approval_handler.predict_tools(sanitized_prompt, envelope)
predicted_tools = prediction["tools"]
tool_confidence = prediction["confidence"]
# Step 10: Handle approval mode
allowed_tools = []
approval_result = None
if approval_mode == "prompt":
# Prompt mode: require user approval
if mock_user_approval:
# Testing mode - mock approval
if mock_user_approval == "approve":
allowed_tools = predicted_tools
elif mock_user_approval == "deny":
return {
"status": "denied",
"message": "User denied execution",
"predicted_tools": predicted_tools
}
else:
# Real mode - format approval prompt
approval_prompt = approval_handler.format_approval_prompt(
predicted_tools,
tool_confidence,
envelope,
prediction.get("reasoning", "")
)
approval_result = {
"status": "awaiting_approval",
"approval_prompt": approval_prompt,
"predicted_tools": predicted_tools,
"confidence": tool_confidence,
"envelope": envelope
}
audit_id, audit_error = _log_audit_event(
audit_logger,
event_type="cortex_approval_requested",
user=os.environ.get("USER", "unknown"),
routing={"decision": route_decision, "confidence": route_confidence},
execution={
"envelope": envelope,
"approval_mode": approval_mode,
"auto_approved": False,
"predicted_tools": predicted_tools,
"allowed_tools": []
},
result={"status": "awaiting_approval"},
security={
"sanitized": sanitize_enabled,
"pii_removed": sanitize_enabled and prompt != sanitized_prompt
}
)
approval_result["audit_id"] = audit_id
approval_result["audit_error"] = audit_error
return approval_result
elif approval_mode == "auto":
# Auto mode: auto-approve all tools
allowed_tools = predicted_tools
elif approval_mode == "envelope_only":
# Envelope only mode: no tool prediction
allowed_tools = None # None means rely on envelope only
# Step 11: Execute with Cortex using the sanitized prompt.
if mock_user_approval:
execution_result = {
"status": "success",
"message": "Execution simulated for mocked approval",
"tools_used": allowed_tools or ["envelope-controlled"],
}
else:
execution_result = execute_cortex_streaming(
prompt=sanitized_prompt,
envelope=envelope_mode,
approval_mode=approval_mode,
allowed_tools=allowed_tools,
timeout_seconds=int(config_manager.get("security.execution_timeout_seconds", 5)),
deploy_confirmed=bool(config_manager.get("security.deploy_envelope_confirmation", True) and envelope_mode == "DEPLOY"),
)
execution_result.setdefault("status", "success" if not execution_result.get("error") else "error")
execution_result.setdefault("tools_used", allowed_tools or ["envelope-controlled"])
# Step 12: Audit logging
audit_id, audit_error = _log_audit_event(
audit_logger,
event_type="cortex_execution",
user=os.environ.get("USER", "unknown"),
routing={"decision": route_decision, "confidence": route_confidence},
execution={
"envelope": envelope,
"approval_mode": approval_mode,
"auto_approved": approval_mode in ["auto", "envelope_only"],
"predicted_tools": predicted_tools,
"allowed_tools": allowed_tools
},
result=execution_result,
security={
"sanitized": sanitize_enabled,
"pii_removed": sanitize_enabled and prompt != sanitized_prompt
}
)
# Step 13: Cache result (optional - for future optimization)
# For now, skip caching
return {
"status": "executed",
"audit_id": audit_id,
"audit_error": audit_error,
"routing": {"decision": route_decision, "confidence": route_confidence},
"approval_mode": approval_mode,
"predicted_tools": predicted_tools,
"allowed_tools": allowed_tools,
"result": execution_result,
"security": {
"sanitized": sanitize_enabled,
"pii_removed": sanitize_enabled and prompt != sanitized_prompt
}
}
def main():
"""CLI entry point for security wrapper."""
parser = argparse.ArgumentParser(
description="Security wrapper for cortex-code skill"
)
parser.add_argument(
"--prompt",
required=True,
help="User prompt to execute"
)
parser.add_argument(
"--config",
help="Path to user config file"
)
parser.add_argument(
"--org-policy",
help="Path to organization policy file"
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Dry-run mode: initialize and validate only"
)
parser.add_argument(
"--envelope",
help="Cortex envelope JSON string"
)
args = parser.parse_args()
# Parse envelope if provided
envelope = None
if args.envelope:
try:
envelope = json.loads(args.envelope)
except json.JSONDecodeError as e:
print(json.dumps({
"status": "error",
"message": f"Invalid envelope JSON: {e}"
}))
sys.exit(1)
# Execute with security
try:
result = execute_with_security(
prompt=args.prompt,
config_path=args.config,
org_policy_path=args.org_policy,
dry_run=args.dry_run,
envelope=envelope
)
print(json.dumps(result, indent=2))
except Exception as e:
print(json.dumps({
"status": "error",
"message": str(e)
}))
sys.exit(1)
if __name__ == "__main__":
main()
security/__init__.py
"""Security layer for cortex-code skill."""
__version__ = "1.0.0"
security/approval_handler.py
#!/usr/bin/env python3
"""
Approval handler for tool prediction and user approval flow.
Predicts which tools Cortex needs and formats approval prompts for users.
"""
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
import sys
from pathlib import Path
# Add scripts directory to path for predict_tools import
scripts_dir = Path(__file__).parent.parent / "scripts"
sys.path.insert(0, str(scripts_dir))
from predict_tools import predict_tools as predict_tools_func
@dataclass
class ApprovalResult:
"""Result of approval process."""
approved: bool
allowed_tools: List[str]
user_response: str
class ApprovalHandler:
"""
Handles tool prediction and user approval flow.
Predicts which tools Cortex needs based on user prompts,
formats approval prompts with confidence scores and warnings,
and parses user responses.
"""
def __init__(self, confidence_threshold: float = 0.7):
"""
Initialize approval handler.
Args:
confidence_threshold: Minimum confidence for predictions (default 0.7)
"""
self.confidence_threshold = confidence_threshold
def predict_tools(self, prompt: str, envelope: Dict[str, Any]) -> Dict[str, Any]:
"""
Predict which tools will be needed for the given prompt.
Args:
prompt: User prompt to analyze
envelope: Request envelope with capabilities and context
Returns:
dict with:
- tools: list of predicted tool names
- confidence: float 0-1 indicating prediction confidence
- reasoning: str explaining the prediction
"""
return predict_tools_func(prompt, envelope)
def format_approval_prompt(
self,
tools: List[str],
confidence: float,
envelope: Dict[str, Any],
reasoning: str
) -> str:
"""
Format approval prompt for user.
Args:
tools: List of predicted tool names
confidence: Prediction confidence (0-1)
envelope: Request envelope with user_prompt and context
reasoning: Explanation of tool prediction
Returns:
Formatted approval prompt string
"""
user_prompt = envelope.get("user_prompt", "Unknown request")
# Build approval prompt
lines = []
lines.append("=" * 70)
lines.append("CORTEX TOOL APPROVAL REQUEST")
lines.append("=" * 70)
lines.append("")
lines.append(f"User Request: {user_prompt}")
lines.append("")
lines.append(f"Predicted Tools ({len(tools)}):")
for tool in tools:
lines.append(f" - {tool}")
lines.append("")
lines.append(f"Prediction Confidence: {confidence:.0%}")
lines.append(f"Reasoning: {reasoning}")
lines.append("")
# Add warning if confidence is below threshold
if confidence < self.confidence_threshold:
lines.append("⚠️ WARNING: Low confidence prediction!")
lines.append(f" Confidence {confidence:.0%} is below threshold {self.confidence_threshold:.0%}")
lines.append(" Tool predictions may be uncertain or incomplete.")
lines.append("")
lines.append("=" * 70)
lines.append("APPROVAL OPTIONS:")
lines.append(" approve - Allow these specific tools for this request")
lines.append(" approve_all - Allow all tools (bypass future approvals)")
lines.append(" deny - Reject this request")
lines.append("=" * 70)
lines.append("")
lines.append("Your response: ")
return "\n".join(lines)
def parse_user_response(self, response: str) -> ApprovalResult:
"""
Parse user response to approval prompt.
Args:
response: User's response string
Returns:
ApprovalResult with approval decision and allowed tools
"""
response_lower = response.strip().lower()
if response_lower == "approve":
return ApprovalResult(
approved=True,
allowed_tools=[], # Will be filled by caller with predicted tools
user_response="approve"
)
elif response_lower == "approve_all":
return ApprovalResult(
approved=True,
allowed_tools=["*"], # Wildcard for all tools
user_response="approve_all"
)
elif response_lower == "deny":
return ApprovalResult(
approved=False,
allowed_tools=[],
user_response="deny"
)
else:
# Unknown response - treat as deny for safety
return ApprovalResult(
approved=False,
allowed_tools=[],
user_response=response
)
security/audit_logger.py
"""Structured JSON audit logging with rotation."""
import hashlib
import json
import os
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Optional
class AuditLogger:
"""Audit logger with structured JSON format and file rotation.
Note: This implementation is designed for single-process use only.
Concurrent writes from multiple processes may result in interleaved
JSON lines or race conditions during rotation. For multi-process
scenarios, consider using a log aggregation service or file locking.
"""
VERSION = "2.0.0"
def __init__(
self,
log_path: Path,
rotation_size: str = "10MB",
retention_days: int = 30
):
"""Initialize audit logger.
Args:
log_path: Path to audit log file
rotation_size: Size threshold for rotation (e.g., "10MB", "1GB")
retention_days: Days to retain rotated logs (NOT YET IMPLEMENTED)
"""
self.log_path = Path(log_path)
self.rotation_size = self._parse_size(rotation_size)
self.retention_days = retention_days
self.initialization_error: Optional[str] = None
# TODO: Implement cleanup of rotated files older than retention_days
try:
self.log_path.parent.mkdir(parents=True, exist_ok=True)
if not self.log_path.exists():
self.log_path.touch(mode=0o600)
else:
os.chmod(self.log_path, 0o600)
except OSError as exc:
self.initialization_error = str(exc)
def log_execution(
self,
event_type: str,
user: str,
routing: Dict[str, Any],
execution: Dict[str, Any],
result: Dict[str, Any],
session_id: Optional[str] = None,
cortex_session_id: Optional[str] = None,
security: Optional[Dict[str, Any]] = None
) -> str:
"""Log a cortex execution event."""
if self.initialization_error:
raise OSError(self.initialization_error)
audit_id = str(uuid.uuid4())
entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"version": self.VERSION,
"audit_id": audit_id,
"event_type": event_type,
"user": user,
"session_id": session_id,
"cortex_session_id": cortex_session_id,
"routing": routing,
"execution": execution,
"result": result,
"security": security or {}
}
entry["prev_hash"] = self._last_entry_hash()
entry["entry_hash"] = self._entry_hash(entry)
self._write_entry(entry)
self._rotate_if_needed()
return audit_id
def _entry_hash(self, entry: Dict[str, Any]) -> str:
"""Hash a canonical audit entry for tamper-evident chaining."""
payload = json.dumps(entry, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(payload.encode()).hexdigest()
def _last_entry_hash(self) -> Optional[str]:
"""Return the previous entry hash if the audit log has entries."""
if not self.log_path.exists():
return None
try:
last_line = None
with open(self.log_path, 'r') as f:
for line in f:
if line.strip():
last_line = line
if not last_line:
return None
return json.loads(last_line).get("entry_hash")
except (OSError, json.JSONDecodeError):
return None
def _write_entry(self, entry: Dict[str, Any]) -> None:
"""Write entry to log file as JSON.
Opens file for each write to avoid holding file handles open long-term.
This trades some efficiency for simplicity and crash-safety (no buffering).
If file was deleted externally, it will be recreated with default permissions.
"""
with open(self.log_path, 'a') as f:
f.write(json.dumps(entry) + '\n')
def _parse_size(self, size_str: str) -> int:
"""Parse size string like '10MB' to bytes."""
size_str = size_str.upper()
multipliers = {
'KB': 1024,
'MB': 1024 * 1024,
'GB': 1024 * 1024 * 1024
}
for suffix, multiplier in multipliers.items():
if size_str.endswith(suffix):
try:
value = float(size_str[:-len(suffix)])
return int(value * multiplier)
except ValueError:
pass
# Default to bytes
try:
return int(size_str)
except ValueError:
return 10 * 1024 * 1024 # Default 10MB
def _rotate_if_needed(self) -> None:
"""Rotate log file if exceeds size limit."""
if not self.log_path.exists():
return
size = self.log_path.stat().st_size
if size >= self.rotation_size:
# Rotate: rename current to .1, .1 to .2, etc.
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
rotated_path = self.log_path.with_suffix(f".{timestamp}.log")
self.log_path.rename(rotated_path)
# Create new log file
self.log_path.touch(mode=0o600)
security/cache_manager.py
"""Secure cache manager with integrity validation."""
import hashlib
import hmac
import json
import os
import time
import warnings
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional
class CacheManager:
"""Secure cache manager with fingerprint validation."""
VERSION = "2.0.0"
def __init__(self, cache_dir: Path):
"""Initialize cache manager."""
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
# Set directory permissions to 0700 (owner only). Some managed or
# sandboxed filesystems deny chmod on existing home-cache directories;
# keep the cache usable rather than failing security-wrapper startup.
try:
os.chmod(self.cache_dir, 0o700)
except PermissionError as exc:
warnings.warn(
f"Could not set secure permissions on cache directory {self.cache_dir}: {exc}",
RuntimeWarning,
stacklevel=2,
)
def _signature_key(self) -> bytes:
"""Return key material for cache tamper detection."""
return os.environ.get(
"CORTEX_CODE_CACHE_HMAC_KEY",
f"cortex-cache:{self.cache_dir}"
).encode()
def _calculate_signature(self, cache_entry: dict) -> str:
"""Calculate HMAC over stable cache fields."""
signed_payload = {
"version": cache_entry.get("version"),
"created_at": cache_entry.get("created_at"),
"expires_at": cache_entry.get("expires_at"),
"data": cache_entry.get("data"),
"fingerprint": cache_entry.get("fingerprint"),
}
payload = json.dumps(signed_payload, sort_keys=True, separators=(",", ":"))
return hmac.new(self._signature_key(), payload.encode(), hashlib.sha256).hexdigest()
def _validate_key(self, key: str) -> None:
"""Validate cache key to prevent path traversal."""
if not key:
raise ValueError("Cache key cannot be empty")
# Allow only alphanumeric, underscore, hyphen, and dot
import re
if not re.match(r'^[a-zA-Z0-9_.-]+$', key):
raise ValueError(
f"Invalid cache key: {key}. "
f"Only alphanumeric characters, underscores, hyphens, and dots are allowed."
)
# Prevent path traversal
if '..' in key or '/' in key or '\\' in key:
raise ValueError(f"Invalid cache key: {key}. Path traversal not allowed.")
def write(self, key: str, data: Any, ttl: int = 86400) -> None:
"""Write data to cache with TTL and fingerprint."""
self._validate_key(key)
cache_entry = {
"version": self.VERSION,
"created_at": datetime.now(timezone.utc).isoformat(),
"expires_at": time.time() + ttl,
"data": data
}
# Calculate fingerprint
data_str = json.dumps(data, sort_keys=True)
fingerprint = hashlib.sha256(data_str.encode()).hexdigest()
cache_entry["fingerprint"] = fingerprint
cache_entry["signature"] = self._calculate_signature(cache_entry)
# Write to file
cache_file = self.cache_dir / f"{key}.json"
with open(cache_file, 'w') as f:
json.dump(cache_entry, f, indent=2)
# Set file permissions to 0600 (owner read/write only)
os.chmod(cache_file, 0o600)
def read(self, key: str) -> Optional[Any]:
"""Read data from cache with validation."""
self._validate_key(key)
cache_file = self.cache_dir / f"{key}.json"
if not cache_file.exists():
return None
try:
with open(cache_file, 'r') as f:
cache_entry = json.load(f)
# Check expiration
if cache_entry["expires_at"] <= time.time():
# Expired - delete and return None
cache_file.unlink(missing_ok=True)
return None
# Validate fingerprint
data = cache_entry["data"]
data_str = json.dumps(data, sort_keys=True)
expected_fingerprint = hashlib.sha256(data_str.encode()).hexdigest()
if cache_entry["fingerprint"] != expected_fingerprint:
# Tampered - delete and return None
cache_file.unlink(missing_ok=True)
return None
expected_signature = self._calculate_signature(cache_entry)
if cache_entry.get("signature") != expected_signature:
# Tampered - delete and return None
cache_file.unlink(missing_ok=True)
return None
return data
except (json.JSONDecodeError, KeyError, FileNotFoundError, OSError):
# Corrupted cache - delete and return None
cache_file.unlink(missing_ok=True)
return None
def clear(self, key: Optional[str] = None) -> None:
"""Clear cache entry or all entries."""
if key:
self._validate_key(key)
cache_file = self.cache_dir / f"{key}.json"
if cache_file.exists():
cache_file.unlink(missing_ok=True)
else:
# Clear all cache files
for cache_file in self.cache_dir.glob("*.json"):
cache_file.unlink(missing_ok=True)
security/config_manager.py
"""Configuration manager with 3-layer precedence."""
import copy
import os
import sys
from pathlib import Path
from typing import Any, Optional, Dict
import yaml
class ConfigValidationError(Exception):
"""Raised when configuration validation fails."""
pass
class ConfigManager:
"""Manages security configuration with precedence: org policy > user config > defaults."""
DEFAULT_CONFIG = {
"security": {
"approval_mode": "prompt",
"tool_prediction_confidence_threshold": 0.7,
"allow_tool_expansion": True,
"audit_log_path": "~/.__CODING_AGENT__/skills/cortex-code/audit.log",
"audit_log_rotation": "10MB",
"audit_log_retention": 30,
"sanitize_conversation_history": True,
"sanitize_session_files": True,
"max_history_items": 3,
"cache_dir": "~/.cache/cortex-skill",
"cache_permissions": "0600",
"allowed_envelopes": ["RO", "RW", "RESEARCH"],
"deploy_envelope_confirmation": True,
"execution_timeout_seconds": 300,
"credential_file_allowlist": [
"~/.ssh/*",
"~/.snowflake/*",
"**/.env",
"**/.env.*",
"**/credentials.json",
"**/*_key.p8",
"**/*_key.pem",
"~/.aws/credentials",
"~/.kube/config"
]
}
}
def __init__(
self,
config_path: Optional[Path] = None,
org_policy_path: Optional[Path] = None
):
"""Initialize config manager."""
self._config = self._load_config(config_path, org_policy_path)
def _validate_config(self, config: Dict) -> None:
"""Validate configuration values."""
security = config.get("security", {})
# Validate approval_mode
approval_mode = security.get("approval_mode")
if approval_mode not in ["prompt", "auto", "envelope_only"]:
raise ConfigValidationError(
f"Invalid approval_mode: {approval_mode}. "
f"Must be one of: prompt, auto, envelope_only"
)
# Validate allowed_envelopes
valid_envelopes = {"RO", "RW", "RESEARCH", "DEPLOY", "NONE"}
allowed_envelopes = security.get("allowed_envelopes", [])
for envelope in allowed_envelopes:
if envelope not in valid_envelopes:
raise ConfigValidationError(
f"Invalid envelope: {envelope}. "
f"Must be one of: {', '.join(valid_envelopes)}"
)
# Validate numeric values
confidence = security.get("tool_prediction_confidence_threshold")
if confidence is not None:
if not isinstance(confidence, (int, float)):
raise ConfigValidationError(
f"tool_prediction_confidence_threshold must be a number, got {type(confidence).__name__}"
)
if not (0 <= confidence <= 1):
raise ConfigValidationError(
f"tool_prediction_confidence_threshold must be between 0 and 1, got {confidence}"
)
retention = security.get("audit_log_retention")
if retention is not None:
if not isinstance(retention, int):
raise ConfigValidationError(
f"audit_log_retention must be an integer, got {type(retention).__name__}"
)
if retention < 0:
raise ConfigValidationError(
f"audit_log_retention must be >= 0, got {retention}"
)
def _safe_placeholder_path(self, original_path: str) -> str:
"""Fallback when install-time __CODING_AGENT__ replacement was not applied."""
suffix = Path(original_path).name or "audit.log"
return str(Path.home() / ".cache" / "cortex-skill" / suffix)
def _expand_paths(self, config: Dict) -> Dict:
"""Expand ~ and environment variables in file paths."""
security = config.get("security", {})
# Expand audit_log_path
if "audit_log_path" in security:
security["audit_log_path"] = os.path.expanduser(security["audit_log_path"])
if "__CODING_AGENT__" in security["audit_log_path"]:
security["audit_log_path"] = self._safe_placeholder_path(security["audit_log_path"])
# Expand cache_dir
if "cache_dir" in security:
security["cache_dir"] = os.path.expanduser(security["cache_dir"])
config["security"] = security
return config
def _load_config(
self,
config_path: Optional[Path],
org_policy_path: Optional[Path]
) -> Dict:
"""Load configuration with 3-layer precedence."""
# Start with defaults
config = copy.deepcopy(self.DEFAULT_CONFIG)
# Load user config if exists
if config_path and config_path.exists():
try:
with open(config_path, 'r') as f:
try:
user_config = yaml.safe_load(f) or {}
config = self._merge_config(config, user_config)
except yaml.YAMLError as e:
print(f"Warning: Failed to parse user config {config_path}: {e}", file=sys.stderr)
except OSError as e:
print(f"Warning: Failed to read user config {config_path}: {e}", file=sys.stderr)
org_policy_security = {}
# Load org policy if exists
if org_policy_path and org_policy_path.exists():
try:
with open(org_policy_path, 'r') as f:
try:
org_policy = yaml.safe_load(f) or {}
org_policy_security = org_policy.get("security", {}) or {}
# If override flag set, org policy wins completely
if org_policy.get("security", {}).get("override_user_config"):
# Merge org policy over defaults (skip user config)
config = self._merge_config(copy.deepcopy(self.DEFAULT_CONFIG), org_policy)
else:
# Normal merge: org policy > user config > defaults
config = self._merge_config(config, org_policy)
except yaml.YAMLError as e:
print(f"Warning: Failed to parse org policy {org_policy_path}: {e}", file=sys.stderr)
except OSError as e:
print(f"Warning: Failed to read org policy {org_policy_path}: {e}", file=sys.stderr)
# Validate before applying floors so invalid user config is still rejected.
self._validate_config(config)
# User config must not relax the security floor unless org policy
# explicitly authorizes the relaxed field/value.
config = self._enforce_security_floor(config, org_policy_security)
# Validate configuration
self._validate_config(config)
# Expand file paths
config = self._expand_paths(config)
return config
def _enforce_security_floor(self, config: Dict, org_policy_security: Optional[Dict] = None) -> Dict:
"""Prevent user config from relaxing defaults without explicit org policy."""
result = copy.deepcopy(config)
security = result.setdefault("security", {})
default_security = self.DEFAULT_CONFIG["security"]
org_policy_security = org_policy_security or {}
if (
security.get("approval_mode") != default_security["approval_mode"]
and "approval_mode" not in org_policy_security
):
security["approval_mode"] = default_security["approval_mode"]
default_envelopes = set(default_security["allowed_envelopes"])
explicit_org_envelopes = set(org_policy_security.get("allowed_envelopes", []))
envelope_floor = default_envelopes | explicit_org_envelopes
requested_envelopes = security.get("allowed_envelopes", default_security["allowed_envelopes"])
security["allowed_envelopes"] = [
envelope for envelope in requested_envelopes
if envelope in envelope_floor
]
return result
def _merge_config(self, base: Dict, override: Dict) -> Dict:
"""Deep merge override into base."""
result = copy.deepcopy(base)
for key, value in override.items():
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = self._merge_config(result[key], value)
else:
result[key] = value
return result
def get(self, key: str, default: Any = None) -> Any:
"""Get config value by dot-notation key."""
keys = key.split(".")
value = self._config
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return default
return value
security/default_policy.yaml
# Default security policy for cortex-code skill
# This file documents the secure defaults - do not modify directly
# To customize, create ~/.claude/skills/cortex-code/config.yaml
security:
# Approval mode: "prompt" | "auto" | "envelope_only"
# Default: "prompt" (most secure - ask user before execution)
approval_mode: "prompt"
# Tool prediction settings (for "prompt" mode)
tool_prediction_confidence_threshold: 0.7
allow_tool_expansion: true
# Audit logging (mandatory when approval_mode: "auto")
audit_log_path: "~/.claude/skills/cortex-code/audit.log"
audit_log_rotation: "10MB"
audit_log_retention: 30
# Prompt sanitization
sanitize_conversation_history: true
sanitize_session_files: true
max_history_items: 3
# Cache security
cache_dir: "~/.cache/cortex-skill"
cache_permissions: "0600"
# Envelope restrictions
allowed_envelopes:
- "RO"
- "RW"
- "RESEARCH"
deploy_envelope_confirmation: true
# Routing security - never route these to Cortex
credential_file_allowlist:
- "~/.ssh/*"
- "~/.snowflake/*"
- "**/.env"
- "**/.env.*"
- "**/credentials.json"
- "**/*_key.p8"
- "**/*_key.pem"
- "~/.aws/credentials"
- "~/.kube/config"
security/policies/default_policy.yaml
# Default security policy for cortex-code skill
# This file documents the secure defaults - do not modify directly
# To customize, create ~/.claude/skills/cortex-code/config.yaml
security:
# Approval mode: "prompt" | "auto" | "envelope_only"
# Default: "prompt" (most secure - ask user before execution)
approval_mode: "prompt"
# Tool prediction settings (for "prompt" mode)
tool_prediction_confidence_threshold: 0.7
allow_tool_expansion: true
# Audit logging (mandatory when approval_mode: "auto")
audit_log_path: "~/.claude/skills/cortex-code/audit.log"
audit_log_rotation: "10MB"
audit_log_retention: 30
# Prompt sanitization
sanitize_conversation_history: true
sanitize_session_files: true
max_history_items: 3
# Cache security
cache_dir: "~/.cache/cortex-skill"
cache_permissions: "0600"
# Envelope restrictions
allowed_envelopes:
- "RO"
- "RW"
- "RESEARCH"
deploy_envelope_confirmation: true
# Routing security - never route these to Cortex
credential_file_allowlist:
- "~/.ssh/*"
- "~/.snowflake/*"
- "**/.env"
- "**/.env.*"
- "**/credentials.json"
- "**/*_key.p8"
- "**/*_key.pem"
- "~/.aws/credentials"
- "~/.kube/config"
security/prompt_sanitizer.py
"""Prompt sanitizer for PII removal and injection detection."""
import re
import unicodedata
from typing import List, Dict, Any
class PromptSanitizer:
"""Sanitizes prompts by removing PII and detecting injection attempts."""
# PII regex patterns
CREDIT_CARD_PATTERN = re.compile(
r'\b(?:\d{4}[-\s]?){3}\d{4}\b' # Matches formats: 1234-5678-9012-3456 or 1234567890123456
)
SSN_PATTERN = re.compile(
r'\b\d{3}-\d{2}-\d{4}\b|' # Matches: 123-45-6789
r'\b\d{9}\b' # Matches: 123456789 (exactly 9 digits)
)
EMAIL_PATTERN = re.compile(
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
)
PHONE_PATTERN = re.compile(
r'\b(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b'
)
API_KEY_PATTERN = re.compile(
r'\b(?:api[_-]?key|token|secret)\s*[:=]\s*["\']?[A-Za-z0-9_./+=-]{8,}["\']?|'
r'\bsk-[A-Za-z0-9_./+=-]{8,}\b|'
r'\b[A-Za-z0-9]{32,}\b',
re.IGNORECASE,
)
ZERO_WIDTH_PATTERN = re.compile(r'[\u200B-\u200D\uFEFF]')
HOMOGLYPH_TRANSLATION = str.maketrans({
'а': 'a', 'А': 'A', # Cyrillic a
'е': 'e', 'Е': 'E', # Cyrillic e
'і': 'i', 'І': 'I', # Cyrillic/Ukrainian i
'о': 'o', 'О': 'O', # Cyrillic o
'р': 'p', 'Р': 'P', # Cyrillic er
'с': 'c', 'С': 'C', # Cyrillic es
'х': 'x', 'Х': 'X', # Cyrillic ha
'у': 'y', 'У': 'Y', # Cyrillic u
})
# Injection detection patterns
INJECTION_PATTERNS = [
re.compile(r'ignore\s+(?:all\s+|the\s+)?(previous|above|prior)\s+(instructions|directions|prompts?)', re.IGNORECASE),
re.compile(r'(enter|enable|activate)\s+developer\s+mode', re.IGNORECASE),
re.compile(r'you\s+are\s+now\s+in\s+developer\s+mode', re.IGNORECASE),
re.compile(r'disregard\s+(?:all\s+|the\s+)?(previous|above|prior)', re.IGNORECASE),
re.compile(r'bypass\s+(restrictions|rules|guidelines)', re.IGNORECASE),
]
def _normalize_for_detection(self, text: str) -> str:
"""Normalize text so obfuscated prompt injections match detection rules."""
normalized = unicodedata.normalize('NFKC', text)
normalized = self.ZERO_WIDTH_PATTERN.sub('', normalized)
normalized = normalized.translate(self.HOMOGLYPH_TRANSLATION)
normalized = ''.join(
char for char in normalized
if unicodedata.category(char) not in {'Cf', 'Mn'}
)
return normalized
def sanitize(self, text: str) -> str:
"""
Sanitize text by removing PII and detecting injection attempts.
Args:
text: The text to sanitize
Returns:
Sanitized text with PII removed and injection warnings added
"""
if not text:
return text
detection_text = self._normalize_for_detection(text)
# Check for injection attempts first
for pattern in self.INJECTION_PATTERNS:
if pattern.search(detection_text):
return "[POTENTIAL INJECTION DETECTED - REMOVED]"
# Remove PII
text = self.CREDIT_CARD_PATTERN.sub('<CREDIT_CARD>', text)
text = self.SSN_PATTERN.sub('<SSN>', text)
text = self.EMAIL_PATTERN.sub('<EMAIL>', text)
text = self.PHONE_PATTERN.sub('<PHONE>', text)
text = self.API_KEY_PATTERN.sub('[API_KEY_REDACTED]', text)
return text
def sanitize_sql_literals(self, sql: str) -> str:
"""
Sanitize SQL string by removing PII from literals.
Args:
sql: The SQL string to sanitize
Returns:
Sanitized SQL string
"""
return self.sanitize(sql)
def sanitize_history(self, history: List[Dict[str, Any]], max_items: int = 3) -> List[Dict[str, Any]]:
"""
Sanitize conversation history by limiting items and removing PII.
Args:
history: List of conversation history items (dicts with 'role' and 'content')
max_items: Maximum number of items to keep (default: 3)
Returns:
Sanitized and limited history list
"""
if not history:
return []
# Keep only the last max_items
limited_history = history[-max_items:] if len(history) > max_items else history
# Sanitize each item's content
sanitized = []
for item in limited_history:
sanitized_item = item.copy()
if 'content' in sanitized_item:
sanitized_item['content'] = self.sanitize(sanitized_item['content'])
sanitized.append(sanitized_item)
return sanitized
SKILL.md
---
name: cortex-code
description: Routes Snowflake-related operations to Cortex Code CLI for specialized Snowflake expertise. Use when user asks about Snowflake databases, data warehouses, SQL queries on Snowflake, Cortex AI features, Snowpark, dynamic tables, data governance in Snowflake, Snowflake security, or mentions "Cortex" explicitly. Do NOT use for general programming, local file operations, non-Snowflake databases, web development, or infrastructure tasks unrelated to Snowflake.
license: Proprietary. See LICENSE for complete terms
metadata:
author: Snowflake Integration Team
version: "1.0.0"
compatibility: Requires Cortex Code CLI installed and configured
---
# Cortex Code Integration Skill
## Install
```bash
# Install via npm skills ecosystem (works with Claude Code, Cursor, Codex, and 40+ agents)
npx skills add snowflake-labs/subagent-cortex-code --copy
# Prerequisite: Cortex Code CLI must be installed and configured
# See: https://docs.snowflake.com/en/user-guide/cortex-code
which cortex # verify installation
```
This skill enables your coding agent to leverage Cortex Code's specialized Snowflake expertise by intelligently routing Snowflake-related operations to Cortex Code CLI in headless mode.
## Architecture Overview
**Routing Principle**: ONLY Snowflake operations → Cortex Code. Everything else → your coding agent.
**Key Components**:
- Dynamic skill discovery at session initialization
- LLM-based semantic routing (not keyword matching)
- Security wrapper with approval modes (prompt/auto/envelope_only)
- Stateless Cortex execution with context enrichment
- Hybrid memory management
- Audit logging for compliance
## Security
The skill includes a security wrapper around Cortex execution with three approval modes:
### Approval Modes
1. **prompt** (default): High security
- User shown approval prompt with predicted tools and confidence
- User must approve before execution
- No audit logging required
- Best for: Interactive sessions, untrusted prompts, production
2. **auto**: Medium security
- All operations auto-approved
- Mandatory audit logging
- Envelopes still enforced
- Best for: Automated workflows, trusted environments
3. **envelope_only**: Medium security
- No tool prediction (faster)
- Auto-approved with audit logging
- Relies on envelope blocklist only
- Best for: Trusted environments, low latency needs
**Configuration**: Set in `config.yaml` in the skill's install directory, or via organization policy.
> **IMPORTANT — `config.yaml` is optional.** The skill ships only `config.yaml.example` as a template. If no `config.yaml` exists, the Python scripts apply safe defaults (`approval_mode: prompt`, `default_envelope: RO`). **Do not search, glob, or `ls` for `config.yaml` before executing** — `ConfigManager` handles this internally. Only read/create `config.yaml` if the user explicitly asks to change settings.
### Built-in Protections
- **Prompt Sanitization**: Automatic PII removal and injection detection
- **Credential Blocking**: Prevents routing when credential paths detected
- **Secure Caching**: SHA256-validated cache in `~/.cache/cortex-skill/`
- **Audit Logging**: Structured JSONL logs (mandatory for auto/envelope_only)
- **Organization Policy**: Enterprise override via `~/.snowflake/cortex/claude-skill-policy.yaml`
## Fast Path for Repeat Queries
**Session state is cached — do not re-run initialization steps on every query.**
Skip the following steps if they've already run in the current session:
- `discover_cortex.py` — output cached to `~/.cache/cortex-skill/cortex-capabilities.json`
- `route_request.py` — for obvious Snowflake queries (user says "Snowflake", "Cortex", "databases", "warehouse", etc.), you can skip routing and go straight to execution
- `cortex connections list` — the active connection doesn't change within a session; reuse it
- Any `config.yaml` / org-policy inspection — `ConfigManager` handles this (see note above)
**Minimal flow for a follow-up Snowflake query** (after the first query in a session):
1. (If `approval_mode: prompt`) ask user for approval
2. Call `execute_cortex.py` with the enriched prompt and envelope
3. Return results
That's it. Three steps — no re-discovery, no re-routing, no config inspection.
## Session Initialization
When this skill is first loaded:
### Step 1: Discover Cortex Capabilities
```bash
PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)
$PYTHON scripts/discover_cortex.py
```
This script:
1. Runs `cortex skill list` to enumerate all available Cortex skills
2. Reads each skill's SKILL.md frontmatter and trigger patterns
3. Caches capabilities with `CacheManager` in the configured cache directory
4. Returns structured data about what Cortex can handle
Expected output: JSON mapping of skill names to their trigger patterns and capabilities.
### Step 2: Load Routing Context
The discovered capabilities are loaded into memory to inform routing decisions throughout the session.
## Workflow: Handling User Requests
### Step 1: Analyze Request with LLM-Based Routing
Before taking any action, analyze the user's request:
```bash
PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)
$PYTHON scripts/route_request.py --prompt "USER_PROMPT_HERE"
```
This script:
1. Loads Cortex capabilities from cache
2. Uses LLM reasoning to classify the request
3. Returns routing decision with confidence score
**Routing Logic**:
- **Route to Cortex** if request involves:
- Snowflake databases, warehouses, schemas, tables
- SQL queries specifically for Snowflake
- Cortex AI features (Cortex Search, Cortex Analyst, ML functions)
- Snowpark, dynamic tables, streams, tasks
- Data governance, data quality, or security in Snowflake context
- User explicitly mentions "Cortex" or "Snowflake"
- **Route to your coding agent** if request involves:
- Local file operations (reading, writing, editing local files)
- General programming (Python, JavaScript, etc. not Snowflake-specific)
- Non-Snowflake databases (PostgreSQL, MySQL, MongoDB, etc.)
- Web development, frontend work
- Infrastructure/DevOps unrelated to Snowflake
- Git operations, GitHub, version control
### Step 2: Execute Based on Routing Decision
#### If routing is `coding_agent` (handle locally):
Handle the request directly using your agent's built-in capabilities. No Cortex involvement.
#### If routed to Cortex Code:
Proceed to Step 3.
### Step 3: Choose Security Envelope and Handle Approval
Before executing Cortex, the security wrapper handles approval based on configured mode.
#### Step 3a: Check Approval Mode
`security_wrapper.py` reads `approval_mode` from `config.yaml` internally — **do not inspect the config file yourself.** If `config.yaml` doesn't exist, the default is `prompt` mode.
- **prompt mode** (default): Requires user approval
- **auto mode**: Auto-approve with audit logging
- **envelope_only mode**: Auto-approve, no tool prediction
#### Step 3b: Handle Approval (if prompt mode)
If using prompt mode:
```bash
PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)
$PYTHON scripts/security_wrapper.py \
--prompt "ENRICHED_PROMPT" \
--envelope "RW"
```
This will:
1. Predict required tools using LLM
2. Display approval prompt to user:
```
Cortex Code needs to execute the following tools:
• snowflake_sql_execute
• Read
• Write
Envelope: RW
Confidence: 85%
Approve execution? [yes/no]
```
3. If approved, proceed to Step 3c
4. If denied, abort execution
#### Step 3c: Determine Security Envelope
Determine the appropriate security envelope based on the operation:
- **RO** (Read-Only): For queries and read operations - blocks Edit, Write, destructive Bash
- **RW** (Read-Write): For data modifications - allows most operations, blocks destructive Bash
- **RESEARCH**: For exploratory work - read access plus web tools
- **DEPLOY**: For deployment operations - blocks destructive Bash commands
- **NONE**: Custom blocklist via --disallowed-tools
### Step 4: Enrich Context for Cortex
Build an enriched prompt that includes:
**Claude Conversation Context**:
- Last 2-3 relevant exchanges from current Claude session
- Any Snowflake-specific details already discussed
**Recent Cortex Session Context**:
```bash
PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)
$PYTHON scripts/read_cortex_sessions.py --limit 3
```
This reads the most recent Cortex session files from `~/.local/share/cortex/sessions/` to understand what Cortex recently worked on.
**Enriched Prompt Format**:
```
# Context from Current Session
[Recent relevant conversation history]
# Recent Cortex Work
[Summary from recent Cortex sessions]
# User Request
[Original user prompt]
```
### Step 5: Execute Cortex Code Headlessly
```bash
PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)
$PYTHON scripts/execute_cortex.py \
--prompt "ENRICHED_PROMPT" \
--connection "connection_name" \
--envelope "RW" \
--disallowed-tools "tool1" "tool2"
```
This script:
1. Invokes `cortex -p "prompt" --output-format stream-json`
2. Uses print mode for prompt delivery and stream JSON output for non-TTY parsing
3. Applies envelope-based security via `--disallowed-tools` blocklist for safety
4. Parses NDJSON event stream in real-time
5. Detects tool use events and execution results
**Key Insight**: The wrapper intentionally does not combine `-p` with `--input-format stream-json`. Cortex reserves `--input-format` for JSON stdin input; with closed stdin, that combination can emit only an init event and exit before processing the prompt.
**Security Envelopes**:
- **RO** (Read-Only): Blocks Edit, Write, destructive Bash commands
- **RW** (Read-Write): Blocks destructive operations like rm -rf, sudo
- **RESEARCH**: Read access plus web tools, blocks write operations
- **DEPLOY**: Deployment operations, blocks destructive Bash commands
- **NONE**: Custom blocklist via --disallowed-tools parameter
**Event Stream Handling**:
- `type: assistant` → Cortex's responses, display to user
- `type: tool_use` → Cortex is calling a tool
- `type: result` → Final outcome
### Step 6: Handle Permission Requests
With the security wrapper:
- **prompt mode**: User approves BEFORE execution (no mid-execution prompts)
- **auto/envelope_only modes**: Non-blocked tools are auto-approved in stream JSON mode
The security wrapper handles permission management through:
1. **Upfront approval** (prompt mode): User approves predicted tools before execution
2. **Audit logging** (auto/envelope_only): All operations logged to `audit.log` in the skill's install directory
3. **Envelope enforcement**: Tool blocklist still enforced via `--disallowed-tools`
### Step 7: Return Results to User
Format Cortex's output for the current session:
- Show SQL query results in readable format
- Display any generated artifacts
- Report success/failure status
- Provide relevant excerpts from Cortex's analysis
## Examples
### Example 1: Snowflake Query
**User says**: "Show me the top 10 customers by revenue in Snowflake"
**Routing**: → Cortex Code (Snowflake SQL query)
**Security Envelope**: RW (allows SQL execution)
**Cortex Action**:
1. Uses snowflake_sql_execute to run: `SELECT customer_name, SUM(revenue) as total FROM sales GROUP BY customer_name ORDER BY total DESC LIMIT 10`
2. Returns formatted results
**Result**: Table displayed to user with top 10 customers.
### Example 2: Local File Operation
**User says**: "Read the config.json file in this directory"
**Routing**: → your coding agent (local file operation)
**Claude Action**: Uses Read tool directly, no Cortex involvement.
**Result**: File contents displayed.
### Example 3: Data Quality Check
**User says**: "Check data quality for the SALES_DATA table"
**Routing**: → Cortex Code (Snowflake data quality - matches Cortex's data-quality skill)
**Security Envelope**: RW (allows SQL execution for analysis)
**Cortex Action**:
1. Runs data quality checks using its data-quality skill
2. Analyzes schema, null rates, duplicates, etc.
3. Generates quality report
**Result**: Comprehensive data quality report with recommendations.
## Important Notes
### Security Wrapper
The skill uses a security wrapper that provides:
- **Approval modes**: prompt (default), auto, envelope_only
- **Prompt sanitization**: Automatic PII removal and injection detection
- **Credential blocking**: Prevents routing when credential paths detected
- **Audit logging**: Mandatory for auto/envelope_only modes
- **Tool prediction**: LLM predicts required tools for approval prompt
**Configuration**: `config.yaml` in the skill's install directory, or via organization policy
### Headless Execution with Auto-Approval
When using auto or envelope_only modes:
- All tool calls are automatically approved without interactive prompts
- Works for built-in tools (Read, Write, Edit, Bash, Grep, Glob) and non-builtin tools (snowflake_sql_execute, data_diff, MCP tools)
- Uses print mode for prompt delivery and stream JSON mode for non-TTY output parsing
- Security is controlled via `--disallowed-tools` blocklist instead of interactive approval; use these modes only in trusted contexts
### Stateless Execution
Each Cortex invocation is stateless. Context must be explicitly provided via enriched prompts.
### Memory Boundaries
- **Your coding agent maintains**: Full conversation history, user preferences, project context
- **Cortex Code receives**: Only task-specific context for current operation
- **Cortex sessions are read**: For historical context enrichment only
### Security Envelope Strategy
Choose envelopes based on operation risk:
1. **Start with RO or RW**: Most operations fit here
2. **Use RESEARCH**: When web access is needed for exploratory work
3. **Use DEPLOY**: Only for deployment-style operations that require broader non-destructive tool access
4. **Use NONE with custom blocklist**: When fine-grained control is needed
### Performance Considerations
- Cortex skill discovery runs once per session (cached)
- Each Cortex execution adds ~2-5 seconds latency
- Use routing wisely to minimize unnecessary Cortex calls
## Troubleshooting
### Error: "Cortex CLI not found"
**Cause**: Cortex Code is not installed or not in PATH
**Solution**:
```bash
which cortex
# If not found, check installation: ~/.snowflake/cortex/
```
### Error: Approval prompt not appearing (or appearing unexpectedly)
**Cause**: Approval mode misconfiguration or organization policy override
**Solution**:
```bash
# Check approval mode (path varies by agent: ~/.claude/, ~/.cursor/, ~/.codex/, etc.)
cat "$(dirname $(which cortex))/../skills/cortex-code/config.yaml" | grep approval_mode 2>/dev/null \
|| cat ~/skills/cortex-code/config.yaml | grep approval_mode
# Check organization policy (overrides user config)
cat ~/.snowflake/cortex/claude-skill-policy.yaml 2>/dev/null
# Expected:
# prompt = shows approval prompts (default)
# auto = auto-approves all operations
# envelope_only = auto-approves, no tool prediction
```
### Error: "Prompt contains credential file path"
**Cause**: Prompt mentions paths matching credential allowlist (e.g., ~/.ssh/, .env)
**Solution**:
1. Remove credential references from prompt
2. Or customize allowlist in config.yaml if false positive
### Error: PII removed from prompts
**Symptom**: Emails, phone numbers replaced with placeholders
**Cause**: Automatic sanitization enabled by default
**Solution**: Disable if needed (not recommended):
```yaml
security:
sanitize_conversation_history: false
```
### Error: "Permission denied" despite auto mode
**Cause**: Tool is in the --disallowed-tools blocklist for current envelope
**Solution**:
1. Check which envelope is being used (RO/RW/RESEARCH/DEPLOY)
2. If operation is safe, switch to a less restrictive envelope
3. Avoid `NONE` in auto/envelope_only modes; use a named envelope plus explicit custom blocklist if needed
### Error: Audit log not created
**Symptom**: No audit.log despite auto/envelope_only mode
**Solution**:
```bash
# Create the skill's install directory if missing and set permissions
# Path is agent-specific: ~/.claude/skills/cortex-code/, ~/.cursor/skills/cortex-code/, etc.
chmod 700 "$(cd "$(dirname "$0")/.." && pwd)"
# Verify audit_log_path in config.yaml within the skill directory
grep audit_log_path config.yaml
```
### Error: Tools still requiring approval
**Cause**: Approval mode, envelope blocklist, or stream JSON invocation is misconfigured
**Solution**: Ensure the wrapper invokes `cortex -p "..." --output-format stream-json` without `--input-format`, and that the configured envelope does not block the intended tool.
### Issue: Routing sends Snowflake query to your coding agent
**Cause**: Routing logic didn't detect Snowflake keywords
**Solution**:
1. Check if user mentioned "Snowflake" explicitly
2. Review routing script logic in `scripts/route_request.py`
3. Add more trigger patterns to routing context
### Issue: Cortex returns "Connection refused"
**Cause**: Snowflake connection not configured in Cortex
**Solution**:
```bash
cortex connections list
# Verify connection is active
# Check ~/.snowflake/cortex/settings.json for cortexAgentConnectionName
```
### Issue: Context enrichment too large
**Cause**: Including too much conversation history
**Solution**: Limit to last 2-3 relevant exchanges, summarize older context.
## Advanced: Custom Routing Rules
To customize routing beyond default logic, edit `scripts/route_request.py`:
```python
# Add custom patterns
FORCE_CORTEX_PATTERNS = [
"snowflake",
"cortex",
"warehouse",
"snowpark"
]
FORCE_CLAUDE_PATTERNS = [
"local file",
"git commit",
"python script" # unless Snowpark
]
```
## References
See `references/` directory for:
- `cortex-cli-reference.md` - Full Cortex CLI documentation
- `routing-examples.md` - More routing decision examples
- `session-file-format.md` - Cortex session file structure
- `troubleshooting-guide.md` - Extended troubleshooting