app.py

#!/usr/bin/env python3
"""ISNAD Scan Web — browser-based security scanner for AI agent skills & MCP configs."""

import json
import os
import shutil
import subprocess
import tempfile
import zipfile
import io
import re
from datetime import datetime, timezone
from pathlib import Path

from flask import Flask, render_template, request, jsonify

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 10 * 1024 * 1024  # 10 MB max upload

ALLOWED_EXTENSIONS = {
    ".py", ".js", ".ts", ".json", ".yaml", ".yml", ".toml", ".md",
    ".txt", ".sh", ".bash", ".cfg", ".ini", ".xml", ".html", ".css",
    ".jsx", ".tsx", ".mjs", ".cjs", ".lock", ".env", ".env.example",
}


def run_scan(scan_dir: str) -> dict:
    """Run isnad-scan on a directory and return parsed results."""
    try:
        result = subprocess.run(
            ["isnad-scan", scan_dir, "--json", "-v"],
            capture_output=True, text=True, timeout=30,
        )
        # isnad-scan outputs JSON to stdout
        try:
            data = json.loads(result.stdout)
        except json.JSONDecodeError:
            data = {"error": "Failed to parse scan output", "raw": result.stdout[:2000]}
        data["exit_code"] = result.returncode
        data["exit_label"] = {0: "SAFE", 1: "CAUTION", 2: "DANGER", 3: "ERROR"}.get(
            result.returncode, "UNKNOWN"
        )
        return data
    except subprocess.TimeoutExpired:
        return {"error": "Scan timed out (30s limit)", "exit_code": 3, "exit_label": "ERROR"}
    except FileNotFoundError:
        return {"error": "isnad-scan not installed on server", "exit_code": 3, "exit_label": "ERROR"}


def save_paste_to_dir(text: str, filename: str, tmp_dir: str):
    """Save pasted code to a file in tmp_dir."""
    if not filename:
        # Try to detect language
        if text.strip().startswith(("{", "[")):
            filename = "config.json"
        elif "import " in text or "def " in text:
            filename = "skill.py"
        elif "function " in text or "const " in text or "require(" in text:
            filename = "skill.js"
        else:
            filename = "skill.txt"
    filepath = os.path.join(tmp_dir, filename)
    with open(filepath, "w") as f:
        f.write(text)


def extract_github_url(url: str) -> tuple[str, str]:
    """Parse GitHub URL into (owner/repo, subpath)."""
    # https://github.com/owner/repo/tree/branch/path
    m = re.match(r"https?://github\.com/([^/]+/[^/]+)(?:/tree/[^/]+/(.*))?", url)
    if m:
        return m.group(1), m.group(2) or ""
    return "", ""


@app.route("/")
def index():
    return render_template("index.html")


@app.route("/scan", methods=["POST"])
def scan():
    tmp_dir = tempfile.mkdtemp(prefix="isnad-scan-")
    try:
        scan_mode = request.form.get("mode", "paste")

        if scan_mode == "paste":
            code = request.form.get("code", "").strip()
            if not code:
                return jsonify({"error": "No code provided"}), 400
            filename = request.form.get("filename", "").strip()
            save_paste_to_dir(code, filename, tmp_dir)

        elif scan_mode == "upload":
            files = request.files.getlist("files")
            if not files or all(f.filename == "" for f in files):
                return jsonify({"error": "No files uploaded"}), 400
            for f in files:
                if not f.filename:
                    continue
                ext = Path(f.filename).suffix.lower()
                if ext == ".zip":
                    # Extract zip
                    zdata = io.BytesIO(f.read())
                    with zipfile.ZipFile(zdata) as zf:
                        for zi in zf.infolist():
                            if zi.is_dir():
                                continue
                            zext = Path(zi.filename).suffix.lower()
                            if zext in ALLOWED_EXTENSIONS:
                                zf.extract(zi, tmp_dir)
                elif ext in ALLOWED_EXTENSIONS:
                    safe_name = Path(f.filename).name
                    f.save(os.path.join(tmp_dir, safe_name))

        elif scan_mode == "url":
            url = request.form.get("url", "").strip()
            if not url:
                return jsonify({"error": "No URL provided"}), 400
            # GitHub repo clone (shallow)
            repo, subpath = extract_github_url(url)
            if repo:
                clone_dir = os.path.join(tmp_dir, "repo")
                result = subprocess.run(
                    ["git", "clone", "--depth", "1", f"https://github.com/{repo}.git", clone_dir],
                    capture_output=True, text=True, timeout=30,
                )
                if result.returncode != 0:
                    return jsonify({"error": f"Git clone failed: {result.stderr[:500]}"}), 400
                if subpath:
                    scan_path = os.path.join(clone_dir, subpath)
                    if not os.path.exists(scan_path):
                        return jsonify({"error": f"Path '{subpath}' not found in repo"}), 400
                    tmp_dir = scan_path  # Point scan at subpath
                else:
                    tmp_dir = clone_dir
            else:
                return jsonify({"error": "Only GitHub URLs supported currently"}), 400
        else:
            return jsonify({"error": "Invalid scan mode"}), 400

        # Check we have files to scan
        file_count = sum(1 for _ in Path(tmp_dir).rglob("*") if _.is_file())
        if file_count == 0:
            return jsonify({"error": "No scannable files found"}), 400

        results = run_scan(tmp_dir)
        results["scanned_at"] = datetime.now(timezone.utc).isoformat()
        results["file_count"] = file_count
        return jsonify(results)

    finally:
        # Clean up temp dir (might have been reassigned for GitHub)
        parent = tmp_dir
        while "/tmp/isnad-scan-" not in parent and parent != "/tmp":
            parent = str(Path(parent).parent)
        if parent.startswith("/tmp/isnad-scan-"):
            shutil.rmtree(parent, ignore_errors=True)


@app.route("/api/scan", methods=["POST"])
def api_scan():
    """JSON API endpoint for programmatic scanning."""
    data = request.get_json()
    if not data:
        return jsonify({"error": "JSON body required"}), 400

    tmp_dir = tempfile.mkdtemp(prefix="isnad-scan-")
    try:
        files = data.get("files", {})
        if not files:
            return jsonify({"error": "Provide 'files' dict: {filename: content}"}), 400

        for fname, content in files.items():
            ext = Path(fname).suffix.lower()
            if ext in ALLOWED_EXTENSIONS or not ext:
                filepath = os.path.join(tmp_dir, Path(fname).name)
                with open(filepath, "w") as f:
                    f.write(content)

        results = run_scan(tmp_dir)
        results["scanned_at"] = datetime.now(timezone.utc).isoformat()
        return jsonify(results)
    finally:
        shutil.rmtree(tmp_dir, ignore_errors=True)


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 8080))
    app.run(host="0.0.0.0", port=port, debug=os.environ.get("DEBUG", "0") == "1")