#!/usr/bin/env python3 """ SAGE Code Scanner — Multi-Layer Security & Quality Analysis Powered by Eden's Intelligence Layers: 1. Pattern matching (instant) 2. AST structural analysis (syntax, complexity, docstrings, type hints) 3. Bandit security scan (CVE-grade Python vulnerabilities) 4. Radon complexity metrics (cyclomatic + maintainability) 5. Eden LLM deep review (ollama — the differentiator) φ = 1.618 | RETIRE_DADDY """ import ast import json import os import re import subprocess import tempfile import time import traceback from datetime import datetime import requests from flask import Flask, request, jsonify, Response app = Flask(__name__) OLLAMA_URL = "http://localhost:11434/api/generate" LLM_MODEL = "eden-coder-omega-v3:latest" LLM_TIMEOUT = 90 PHI = 1.618033988749895 # ═══════════════════════════════════════════════════════════ # Layer 1: Pattern Matching (all languages) # ═══════════════════════════════════════════════════════════ SECURITY_PATTERNS = { # Injection "eval(": {"risk": "Code Injection", "severity": "HIGH", "cwe": "CWE-94"}, "exec(": {"risk": "Code Injection", "severity": "HIGH", "cwe": "CWE-94"}, "os.system": {"risk": "Command Injection", "severity": "CRITICAL", "cwe": "CWE-78"}, "subprocess.call": {"risk": "Command Execution", "severity": "HIGH", "cwe": "CWE-78"}, "shell=True": {"risk": "Shell Injection", "severity": "CRITICAL", "cwe": "CWE-78"}, "__import__": {"risk": "Dynamic Import", "severity": "MEDIUM", "cwe": "CWE-502"}, # Deserialization "pickle.loads": {"risk": "Unsafe Deserialization", "severity": "HIGH", "cwe": "CWE-502"}, "yaml.load(": {"risk": "Unsafe YAML Load", "severity": "HIGH", "cwe": "CWE-502"}, "marshal.loads": {"risk": "Unsafe Deserialization", "severity": "HIGH", "cwe": "CWE-502"}, # Crypto "md5": {"risk": "Weak Hash (MD5)", "severity": "MEDIUM", "cwe": "CWE-328"}, "sha1": {"risk": "Weak Hash (SHA1)", "severity": "LOW", "cwe": "CWE-328"}, "DES": {"risk": "Weak Cipher (DES)", "severity": "HIGH", "cwe": "CWE-327"}, # SQL "execute(\"": {"risk": "Possible SQL Injection", "severity": "HIGH", "cwe": "CWE-89"}, "execute('": {"risk": "Possible SQL Injection", "severity": "HIGH", "cwe": "CWE-89"}, "f\"SELECT": {"risk": "SQL Injection via f-string", "severity": "CRITICAL", "cwe": "CWE-89"}, "f'SELECT": {"risk": "SQL Injection via f-string", "severity": "CRITICAL", "cwe": "CWE-89"}, # Secrets "password =": {"risk": "Hardcoded Password", "severity": "MEDIUM", "cwe": "CWE-798"}, "api_key =": {"risk": "Hardcoded API Key", "severity": "MEDIUM", "cwe": "CWE-798"}, "secret =": {"risk": "Hardcoded Secret", "severity": "MEDIUM", "cwe": "CWE-798"}, # Web "mark_safe": {"risk": "XSS via mark_safe", "severity": "HIGH", "cwe": "CWE-79"}, "| safe": {"risk": "XSS via safe filter", "severity": "HIGH", "cwe": "CWE-79"}, "innerHTML": {"risk": "XSS via innerHTML", "severity": "HIGH", "cwe": "CWE-79"}, # File "open(": {"risk": "File Operation (check path validation)", "severity": "INFO", "cwe": "CWE-22"}, # Rust/systems "unsafe {": {"risk": "Unsafe Rust Block", "severity": "MEDIUM", "cwe": "CWE-119"}, "saturating_add": {"risk": "Potential Gas Metering Bypass", "severity": "HIGH", "cwe": "N/A"}, "unchecked_math": {"risk": "Arithmetic Overflow Risk", "severity": "HIGH", "cwe": "CWE-190"}, } def layer1_patterns(code): """Fast pattern scan across all languages.""" findings = [] lines = code.split("\n") for i, line in enumerate(lines, 1): stripped = line.strip() if stripped.startswith("#") or stripped.startswith("//"): continue for pattern, info in SECURITY_PATTERNS.items(): if pattern in line: if info["severity"] == "INFO": continue findings.append({ "line": i, "pattern": pattern, "risk": info["risk"], "severity": info["severity"], "cwe": info["cwe"], "code": stripped[:120], "layer": "pattern_match" }) return findings # ═══════════════════════════════════════════════════════════ # Layer 2: AST Analysis (Python only) # ═══════════════════════════════════════════════════════════ def layer2_ast(code, lang): """Structural analysis via AST (Python only).""" if lang != "python": return {"available": False, "reason": f"AST analysis not available for {lang}"} result = { "available": True, "syntax_valid": False, "functions": 0, "classes": 0, "imports": 0, "lines": len(code.split("\n")), "issues": [], "suggestions": [], } try: tree = ast.parse(code) result["syntax_valid"] = True except SyntaxError as e: result["issues"].append({ "type": "syntax_error", "message": f"{e.msg} at line {e.lineno}", "severity": "CRITICAL", "line": e.lineno }) return result funcs = [n for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))] classes = [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)] imports = [n for n in ast.walk(tree) if isinstance(n, (ast.Import, ast.ImportFrom))] result["functions"] = len(funcs) result["classes"] = len(classes) result["imports"] = len(imports) # Missing docstrings missing_docs = [] for func in funcs: if not ast.get_docstring(func): missing_docs.append(func.name) if missing_docs: result["suggestions"].append({ "type": "missing_docstrings", "functions": missing_docs[:10], "count": len(missing_docs) }) # Missing type hints missing_hints = [] for func in funcs: has_hints = func.returns is not None or any( arg.annotation is not None for arg in func.args.args ) if not has_hints: missing_hints.append(func.name) if missing_hints: result["suggestions"].append({ "type": "missing_type_hints", "functions": missing_hints[:10], "count": len(missing_hints) }) # Long functions for func in funcs: try: func_code = ast.unparse(func) func_lines = func_code.count("\n") + 1 if func_lines > 40: result["issues"].append({ "type": "long_function", "function": func.name, "lines": func_lines, "severity": "LOW", "message": f"Function '{func.name}' is {func_lines} lines — consider refactoring" }) except: pass # Global variables global_assigns = [n for n in ast.iter_child_nodes(tree) if isinstance(n, ast.Assign)] if len(global_assigns) > 10: result["issues"].append({ "type": "excessive_globals", "count": len(global_assigns), "severity": "LOW", "message": f"{len(global_assigns)} global assignments — consider using classes or modules" }) # Bare except for node in ast.walk(tree): if isinstance(node, ast.ExceptHandler) and node.type is None: result["issues"].append({ "type": "bare_except", "line": node.lineno, "severity": "MEDIUM", "message": f"Bare 'except:' at line {node.lineno} — catches all exceptions including KeyboardInterrupt" }) return result # ═══════════════════════════════════════════════════════════ # Layer 3: Bandit Security Scan (Python only) # ═══════════════════════════════════════════════════════════ def layer3_bandit(code, lang): """Run bandit security scanner.""" if lang != "python": return {"available": False, "reason": f"Bandit only supports Python"} try: with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: f.write(code) tmp_path = f.name result = subprocess.run( ["bandit", "-f", "json", "-q", tmp_path], capture_output=True, text=True, timeout=30 ) os.unlink(tmp_path) if result.stdout: data = json.loads(result.stdout) findings = [] for r in data.get("results", []): findings.append({ "test_id": r.get("test_id", ""), "test_name": r.get("test_name", ""), "severity": r.get("issue_severity", "UNKNOWN"), "confidence": r.get("issue_confidence", "UNKNOWN"), "risk": r.get("issue_text", ""), "line": r.get("line_number", 0), "cwe": f"CWE-{r['issue_cwe']['id']}" if r.get("issue_cwe") else "N/A", "code": r.get("code", "").strip()[:200], "layer": "bandit" }) metrics = data.get("metrics", {}).get(tmp_path, data.get("metrics", {}).get("_totals", {})) return { "available": True, "findings": findings, "severity_counts": { "high": metrics.get("SEVERITY.HIGH", 0), "medium": metrics.get("SEVERITY.MEDIUM", 0), "low": metrics.get("SEVERITY.LOW", 0), }, "confidence_counts": { "high": metrics.get("CONFIDENCE.HIGH", 0), "medium": metrics.get("CONFIDENCE.MEDIUM", 0), "low": metrics.get("CONFIDENCE.LOW", 0), } } return {"available": True, "findings": [], "severity_counts": {}, "confidence_counts": {}} except subprocess.TimeoutExpired: try: os.unlink(tmp_path) except: pass return {"available": False, "reason": "Bandit timed out"} except Exception as e: try: os.unlink(tmp_path) except: pass return {"available": False, "reason": str(e)} # ═══════════════════════════════════════════════════════════ # Layer 4: Radon Complexity Analysis (Python only) # ═══════════════════════════════════════════════════════════ def layer4_radon(code, lang): """Run radon cyclomatic complexity + maintainability analysis.""" if lang != "python": return {"available": False, "reason": f"Radon only supports Python"} try: with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: f.write(code) tmp_path = f.name # Cyclomatic complexity cc_result = subprocess.run( ["radon", "cc", "-j", "-a", tmp_path], capture_output=True, text=True, timeout=15 ) # Maintainability index mi_result = subprocess.run( ["radon", "mi", "-j", tmp_path], capture_output=True, text=True, timeout=15 ) os.unlink(tmp_path) cc_data = {} mi_data = {} try: cc_data = json.loads(cc_result.stdout) if cc_result.stdout else {} except: pass try: mi_data = json.loads(mi_result.stdout) if mi_result.stdout else {} except: pass functions = [] for path, items in cc_data.items(): if isinstance(items, list): for item in items: if isinstance(item, dict): functions.append({ "name": item.get("name", "?"), "type": item.get("type", "?"), "complexity": item.get("complexity", 0), "rank": item.get("rank", "?"), "line": item.get("lineno", 0), }) mi_score = None mi_rank = None for path, score in mi_data.items(): if isinstance(score, dict): mi_score = score.get("mi", None) mi_rank = score.get("rank", None) elif isinstance(score, (int, float)): mi_score = score complex_funcs = [f for f in functions if f["complexity"] > 10] return { "available": True, "maintainability_index": round(mi_score, 1) if mi_score else None, "maintainability_rank": mi_rank, "functions_analyzed": len(functions), "complex_functions": complex_funcs, "avg_complexity": round( sum(f["complexity"] for f in functions) / max(len(functions), 1), 1 ), "max_complexity": max((f["complexity"] for f in functions), default=0), } except Exception as e: try: os.unlink(tmp_path) except: pass return {"available": False, "reason": str(e)} # ═══════════════════════════════════════════════════════════ # Layer 5: Eden LLM Deep Review # ═══════════════════════════════════════════════════════════ def layer5_llm(code, lang, findings_summary=""): """Use Eden's LLM for deep code review.""" prompt = f"""You are SAGE, a senior code security and quality auditor. Review this {lang} code. Previous automated scans found: {findings_summary or 'nothing notable'} Provide a focused review covering: 1. Security vulnerabilities the pattern scanner might have missed 2. Logic errors or edge cases 3. Performance concerns 4. Architecture suggestions 5. One specific, actionable improvement with a code example Be concise and specific. No generic advice. Reference actual line content. ```{lang} {code[:4000]} ``` REVIEW:""" try: resp = requests.post(OLLAMA_URL, json={ "model": LLM_MODEL, "prompt": prompt, "stream": False, "options": {"num_predict": 800, "temperature": 0.3} }, timeout=LLM_TIMEOUT) if resp.ok: text = resp.json().get("response", "").strip() # Clean response text = re.sub(r".*?", "", text, flags=re.DOTALL).strip() return {"available": True, "review": text} return {"available": False, "reason": f"Ollama returned {resp.status_code}"} except requests.Timeout: return {"available": False, "reason": "LLM review timed out (90s)"} except Exception as e: return {"available": False, "reason": str(e)} # ═══════════════════════════════════════════════════════════ # Grading # ═══════════════════════════════════════════════════════════ def compute_grade(pattern_findings, bandit_data, ast_data, radon_data): """Compute overall grade A-F.""" score = 100 # Pattern findings for f in pattern_findings: if f["severity"] == "CRITICAL": score -= 20 elif f["severity"] == "HIGH": score -= 12 elif f["severity"] == "MEDIUM": score -= 6 # Bandit findings if bandit_data.get("available"): for f in bandit_data.get("findings", []): if f["severity"] == "HIGH": score -= 15 elif f["severity"] == "MEDIUM": score -= 8 elif f["severity"] == "LOW": score -= 3 # AST issues if ast_data.get("available"): for iss in ast_data.get("issues", []): if iss["severity"] == "CRITICAL": score -= 20 elif iss["severity"] == "MEDIUM": score -= 5 elif iss["severity"] == "LOW": score -= 2 # Radon complexity if radon_data.get("available"): mi = radon_data.get("maintainability_index") if mi is not None: if mi < 10: score -= 20 elif mi < 20: score -= 10 if radon_data.get("max_complexity", 0) > 20: score -= 10 score = max(0, min(100, score)) if score >= 90: grade = "A" elif score >= 75: grade = "B" elif score >= 60: grade = "C" elif score >= 40: grade = "D" else: grade = "F" return grade, score # ═══════════════════════════════════════════════════════════ # API Routes # ═══════════════════════════════════════════════════════════ @app.route("/health") def health(): return jsonify({ "status": "SAGE Online", "version": "3.0", "layers": ["pattern_match", "ast_analysis", "bandit", "radon", "eden_llm"], "timestamp": datetime.now().isoformat() }) @app.route("/api/review", methods=["POST"]) def api_review(): """Quick review (layers 1-4, no LLM).""" data = request.json or {} code = data.get("code", "") lang = data.get("lang", "python").lower() if not code: return jsonify({"error": "No code provided"}), 400 if len(code) > 100_000: return jsonify({"error": "Code too large (max 100KB)"}), 400 t0 = time.time() patterns = layer1_patterns(code) ast_data = layer2_ast(code, lang) bandit_data = layer3_bandit(code, lang) radon_data = layer4_radon(code, lang) grade, score = compute_grade(patterns, bandit_data, ast_data, radon_data) return jsonify({ "status": "success", "scan_time_ms": round((time.time() - t0) * 1000), "grade": grade, "score": score, "language": lang, "lines": len(code.split("\n")), "layers": { "pattern_match": {"findings": patterns}, "ast_analysis": ast_data, "bandit": bandit_data, "radon": radon_data, } }) @app.route("/api/deep-review", methods=["POST"]) def api_deep_review(): """Full review including LLM analysis (layers 1-5).""" data = request.json or {} code = data.get("code", "") lang = data.get("lang", "python").lower() if not code: return jsonify({"error": "No code provided"}), 400 if len(code) > 100_000: return jsonify({"error": "Code too large (max 100KB)"}), 400 t0 = time.time() patterns = layer1_patterns(code) ast_data = layer2_ast(code, lang) bandit_data = layer3_bandit(code, lang) radon_data = layer4_radon(code, lang) grade, score = compute_grade(patterns, bandit_data, ast_data, radon_data) # Build summary for LLM summary_parts = [] if patterns: summary_parts.append(f"{len(patterns)} pattern matches ({', '.join(set(f['risk'] for f in patterns[:5]))})") if bandit_data.get("findings"): summary_parts.append(f"{len(bandit_data['findings'])} bandit findings") if radon_data.get("complex_functions"): summary_parts.append(f"{len(radon_data['complex_functions'])} high-complexity functions") summary = "; ".join(summary_parts) if summary_parts else "no automated findings" llm_data = layer5_llm(code, lang, summary) return jsonify({ "status": "success", "scan_time_ms": round((time.time() - t0) * 1000), "grade": grade, "score": score, "language": lang, "lines": len(code.split("\n")), "layers": { "pattern_match": {"findings": patterns}, "ast_analysis": ast_data, "bandit": bandit_data, "radon": radon_data, "eden_llm": llm_data, } }) # ═══════════════════════════════════════════════════════════ # Landing Page # ═══════════════════════════════════════════════════════════ LANDING_HTML = """ SAGE — AI Code Security Scanner
AI-POWERED

5-Layer Code Security Scanner

Paste your code. Get a security audit powered by static analysis, AST parsing, Bandit CVE detection, complexity metrics, and Eden's AI review.

Pattern Match
AST Analysis
Bandit CVE
Radon Complexity
Eden AI Review

// SAGE Products

Single File Review

$25

Security, performance & quality analysis of one file

PR Review

$50

Full pull request analysis with line-by-line comments

Security Scan

$100

OWASP Top 10 vulnerability scan + report

Full Repo Audit

$200

Complete codebase — architecture, security, tech debt

Contact: jameyecho@gmail.com | PayPal: paypal.me/jamlen

SAGE v3.0 — Powered by Eden | scanner.edensages.org
""" @app.route("/") def home(): return Response(LANDING_HTML, mimetype="text/html") # ═══════════════════════════════════════════════════════════ # Backwards-compatible /api/audit endpoint # ═══════════════════════════════════════════════════════════ @app.route("/api/audit", methods=["POST"]) def audit_compat(): """Legacy endpoint — redirects to new review.""" data = request.json or {} code = data.get("code", "") lang = data.get("language", "python").lower() if not code: return jsonify({"error": "No code provided"}), 400 patterns = layer1_patterns(code) grade, score = compute_grade(patterns, {}, {}, {}) return jsonify({ "status": "success", "report": { "grade": grade, "score": score, "vulnerabilities": patterns, } }) if __name__ == "__main__": print("SAGE v3.0 — 5-Layer Code Security Scanner") print(f" Layers: pattern_match, ast, bandit, radon, eden_llm") print(f" Model: {LLM_MODEL}") print(f" Port: 8080") app.run(host="0.0.0.0", port=8080)