# Common shadow AI pattern: developer uses personal API key
# in production code
 
import openai
 
# Personal API key hardcoded or in local .env
openai.api_key = "sk-personal-key-here"
 
def process_customer_ticket(ticket_text):
    """Uses AI to classify and route support tickets.
    Deployed by a single developer without security review."""
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{
            "role": "user",
            "content": f"Classify this support ticket: {ticket_text}"
        }]
    )
    return response.choices[0].message.content

class ShadowAINetworkDetector:
    """Detect shadow AI by monitoring network traffic."""
 
    AI_API_DOMAINS = [
        "api.openai.com",
        "api.anthropic.com",
        "generativelanguage.googleapis.com",
        "api.mistral.ai",
        "api.cohere.ai",
        "api.together.xyz",
        "api.fireworks.ai",
        "api.replicate.com",
        "api.huggingface.co",
    ]
 
    def analyze_dns_logs(self, dns_log_entries):
        """Identify DNS queries to known AI API endpoints."""
        findings = []
 
        for entry in dns_log_entries:
            for domain in self.AI_API_DOMAINS:
                if domain in entry["query"]:
                    findings.append({
                        "domain": domain,
                        "source_ip": entry["source_ip"],
                        "timestamp": entry["timestamp"],
                        "query_count": entry.get("count", 1)
                    })
 
        return self.deduplicate_and_rank(findings)
 
    def analyze_proxy_logs(self, proxy_log_entries):
        """Analyze HTTP proxy logs for AI API traffic."""
        ai_traffic = []
 
        for entry in proxy_log_entries:
            url = entry.get("url", "")
            for domain in self.AI_API_DOMAINS:
                if domain in url:
                    ai_traffic.append({
                        "url": url,
                        "method": entry.get("method"),
                        "user": entry.get("user"),
                        "source_ip": entry.get("source_ip"),
                        "content_type": entry.get("content_type"),
                        "request_size": entry.get("request_size"),
                        "timestamp": entry.get("timestamp")
                    })
 
        return ai_traffic

class ShadowAICodeScanner:
    """Scan code repositories for shadow AI usage."""
 
    INDICATORS = {
        "api_key_patterns": [
            r"sk-[a-zA-Z0-9]{32,}",       # OpenAI key format
            r"sk-ant-[a-zA-Z0-9-]{32,}",   # Anthropic key format
            r"AIza[a-zA-Z0-9_-]{35}",       # Google API key
        ],
        "import_patterns": [
            r"import openai",
            r"from anthropic import",
            r"import google.generativeai",
            r"from langchain",
            r"import llama_index",
            r"from transformers import",
        ],
        "api_url_patterns": [
            r"api\.openai\.com",
            r"api\.anthropic\.com",
            r"generativelanguage\.googleapis\.com",
        ]
    }
 
    def scan_repository(self, repo_path):
        """Scan a code repository for shadow AI indicators."""
        findings = []
 
        for root, dirs, files in os.walk(repo_path):
            # Skip common non-code directories
            dirs[:] = [d for d in dirs if d not in
                       ['.git', 'node_modules', '__pycache__', 'venv']]
 
            for filename in files:
                if filename.endswith(('.py', '.js', '.ts', '.go', '.java',
                                     '.env', '.yaml', '.yml', '.json')):
                    filepath = os.path.join(root, filename)
                    file_findings = self.scan_file(filepath)
                    findings.extend(file_findings)
 
        return findings

def assess_shadow_ai_risk(deployment):
    """Assess the risk of a discovered shadow AI deployment."""
    risk_factors = {
        "data_sensitivity": {
            "pii": 10,
            "financial": 9,
            "healthcare": 10,
            "internal_only": 5,
            "public": 1
        },
        "data_volume": {
            "high": 8,    # Thousands of records/day
            "medium": 5,  # Hundreds of records/day
            "low": 2      # Occasional use
        },
        "authorization": {
            "none": 10,        # No approval at all
            "informal": 7,     # Manager approved, no security review
            "partial": 4,      # Some review but incomplete
            "approved": 1      # Fully approved (not shadow AI)
        },
        "controls": {
            "none": 10,
            "basic": 6,
            "moderate": 3,
            "comprehensive": 1
        }
    }
 
    total_risk = sum(
        risk_factors[category].get(deployment.get(category, "none"), 5)
        for category in risk_factors
    )
    max_risk = sum(max(v.values()) for v in risk_factors.values())
 
    return {
        "risk_score": total_risk / max_risk,
        "risk_level": (
            "critical" if total_risk / max_risk > 0.8 else
            "high" if total_risk / max_risk > 0.6 else
            "medium" if total_risk / max_risk > 0.4 else
            "low"
        ),
        "factors": deployment
    }