# app.py — MetaGuardSOT Web Interface from __future__ import annotations import os import sys import shutil import tempfile from dataclasses import asdict from typing import List from flask import Flask, render_template, request, jsonify, send_file from core import scan_target, ScanResult, sanitize_file from reporters import export_json, export_csv, export_html app = Flask(__name__) app.config["MAX_CONTENT_LENGTH"] = 100 * 1024 * 1024 # 100 MB app.config["UPLOAD_FOLDER"] = os.path.join(os.path.dirname(__file__), "uploads") os.makedirs(app.config["UPLOAD_FOLDER"], exist_ok=True) results_store: List[ScanResult] = [] @app.route("/") def index(): return render_template("index.html") @app.route("/api/upload", methods=["POST"]) def upload_files(): files = request.files.getlist("files") if not files or all(f.filename == "" for f in files): return jsonify({"error": "No files uploaded"}), 400 saved = [] upload_dir = app.config["UPLOAD_FOLDER"] for f in files: if f.filename: safe_name = f.filename.replace("/", "_").replace("\\", "_") dest = os.path.join(upload_dir, safe_name) f.save(dest) saved.append(safe_name) return jsonify({"files": saved, "count": len(saved)}) @app.route("/api/scan", methods=["POST"]) def scan(): data = request.get_json() target = data.get("target", "") recursive = data.get("recursive", True) context = data.get("context", "OSINT") if not target: upload_dir = app.config["UPLOAD_FOLDER"] if os.path.isdir(upload_dir) and os.listdir(upload_dir): target = upload_dir else: return jsonify({"error": "No files to scan. Upload files first."}), 400 try: results = scan_target(target, recursive, context) global results_store results_store = results out = [] for r in results: out.append({ "path": r.path, "file_type": r.file_type, "size_bytes": r.size_bytes, "sha256": r.sha256, "metadata": r.metadata, "context": r.context, "total_risk": r.total_risk, "level": r.level, "findings": [{"name": f.name, "severity": f.severity, "evidence": f.evidence} for f in r.findings], }) return jsonify({"results": out, "count": len(out)}) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/results") def get_results(): out = [] for r in results_store: out.append({ "path": r.path, "file_type": r.file_type, "size_bytes": r.size_bytes, "sha256": r.sha256, "metadata": r.metadata, "context": r.context, "total_risk": r.total_risk, "level": r.level, "findings": [{"name": f.name, "severity": f.severity, "evidence": f.evidence} for f in r.findings], }) return jsonify({"results": out, "count": len(out)}) @app.route("/api/result/") def get_result(index): if 0 <= index < len(results_store): r = results_store[index] return jsonify({ "path": r.path, "file_type": r.file_type, "size_bytes": r.size_bytes, "sha256": r.sha256, "metadata": r.metadata, "context": r.context, "total_risk": r.total_risk, "level": r.level, "findings": [{"name": f.name, "severity": f.severity, "evidence": f.evidence} for f in r.findings], }) return jsonify({"error": "Not found"}), 404 @app.route("/api/sanitize/", methods=["POST"]) def sanitize(index): if 0 <= index < len(results_store): r = results_store[index] if r.file_type != "image": return jsonify({"error": "Sanitization supported only for image files"}), 400 try: clean_path = sanitize_file(r.path, r.file_type) return jsonify({"sanitized_path": clean_path}) except Exception as e: return jsonify({"error": str(e)}), 500 return jsonify({"error": "Not found"}), 404 @app.route("/api/export/") def export(fmt): if not results_store: return jsonify({"error": "No results to export. Run scan first."}), 400 tmp = tempfile.NamedTemporaryFile(delete=False, suffix=f".{fmt}") tmp.close() try: if fmt == "json": export_json(results_store, tmp.name) elif fmt == "csv": export_csv(results_store, tmp.name) elif fmt == "html": export_html(results_store, tmp.name) else: return jsonify({"error": f"Unknown format: {fmt}"}), 400 return send_file(tmp.name, as_attachment=True, download_name=f"report.{fmt}") except Exception as e: return jsonify({"error": str(e)}), 500 finally: try: os.unlink(tmp.name) except OSError: pass @app.route("/api/clear", methods=["POST"]) def clear(): global results_store results_store = [] upload_dir = app.config["UPLOAD_FOLDER"] if os.path.isdir(upload_dir): shutil.rmtree(upload_dir) os.makedirs(upload_dir, exist_ok=True) return jsonify({"status": "cleared"}) if __name__ == "__main__": port = int(os.environ.get("PORT", 5000)) debug = os.environ.get("DEBUG", "1") == "1" print(f"MetaGuardSOT starting on http://127.0.0.1:{port}") app.run(host="0.0.0.0", port=port, debug=debug)