# app/blueprints/api.py """ API blueprint for JSON endpoints. Endpoints: POST /api/analyze_script Body: { "job_id": "", # or "uuid": "" "url": "https://cdn.example.com/app.js", "category": "script" # optional, defaults to "script" } Response: { "ok": true, "final_url": "...", "status_code": 200, "bytes": 12345, "truncated": false, "sha256": "...", "artifact_path": "/abs/path/to//scripts/fetched/.js", "findings": [ { "name": "...", "description": "...", "severity": "...", "tags": [...], "reason": "..." }, ... ], "snippet": "", "snippet_len": 45678 } """ import os import time from flask import Blueprint, request, jsonify, current_app, send_file, abort from pathlib import Path from app.logging_setup import get_app_logger from app.utils.settings import get_settings from app.utils.external_fetcher import ExternalScriptFetcher from werkzeug.exceptions import HTTPException api_bp = Blueprint("api", __name__, url_prefix="/api") app_logger = get_app_logger() def _resolve_results_path(job_id: str) -> str: """ Compute the absolute results directory for a given job UUID. Prefers /artifacts/, falls back to /. """ base_dir = "/data" candidate_with_artifacts = os.path.join(base_dir, job_id) if os.path.isdir(candidate_with_artifacts): return candidate_with_artifacts fallback = os.path.join(base_dir, job_id) os.makedirs(fallback, exist_ok=True) return fallback def _make_snippet(text: str, max_chars: int = 1200) -> str: """Produce a trimmed, safe-to-render snippet of the script contents.""" if not text: return "" snippet = text.strip() return (snippet[:max_chars] + "…") if len(snippet) > max_chars else snippet @api_bp.errorhandler(400) @api_bp.errorhandler(403) @api_bp.errorhandler(404) @api_bp.errorhandler(405) def _api_err(err): """ Return JSON for common client errors. """ if isinstance(err, HTTPException): code = err.code name = (err.name or "error").lower() else: code = 400 name = "error" return jsonify({"ok": False, "error": name}), code @api_bp.errorhandler(500) def _api_500(err): """ Return JSON for server errors and log the exception. """ try: app_logger.exception("API 500") except Exception: pass return jsonify({"ok": False, "error": "internal server error"}), 500 @api_bp.post("/analyze_script") def analyze_script(): """ Analyze EXACTLY one external script URL for a given job UUID. Expected JSON body: { "job_id": "", "url": "https://cdn.example.com/app.js", "category": "script" } """ body = request.get_json(silent=True) or {} job_id_raw = body.get("job_id") or body.get("uuid") script_url_raw = body.get("url") category = (body.get("category") or "script").strip() or None # default to "script" job_id = (job_id_raw or "").strip() if isinstance(job_id_raw, str) else "" script_url = (script_url_raw or "").strip() if isinstance(script_url_raw, str) else "" # log this request app_logger.info(f"Got request to analyze {script_url} via API ") if not job_id or not script_url: return jsonify({"ok": False, "error": "Missing job_id (or uuid) or url"}), 400 settings = get_settings() if not settings.external_fetch.enabled: return jsonify({"ok": False, "error": "Feature disabled"}), 400 # Resolve the UUID-backed results directory for this run. results_path = _resolve_results_path(job_id) # Initialize the fetcher; it reads its own settings internally. fetcher = ExternalScriptFetcher(results_path=results_path) # Unique index for the saved file name: /scripts/fetched/.js unique_index = int(time.time() * 1000) outcome = fetcher.fetch_one(script_url=script_url, index=unique_index) if not outcome.ok or not outcome.saved_path: return jsonify({ "ok": False, "error": outcome.reason, "status_code": outcome.status_code, "final_url": outcome.final_url }), 502 # Read bytes and decode to UTF-8 for rules and snippet try: with open(outcome.saved_path, "rb") as fh: js_text = fh.read().decode("utf-8", errors="ignore") except Exception: js_text = "" # Pull the rules engine from the app (prefer attribute, then config). findings = [] try: engine = getattr(current_app, "rule_engine", None) if engine is None: engine = current_app.config.get("RULE_ENGINE") except Exception: engine = None if engine is not None and hasattr(engine, "run_all"): try: # run_all returns PASS/FAIL for each rule; we only surface FAIL (matched) to the UI all_results = engine.run_all(js_text, category=category) if isinstance(all_results, list): matched = [] for r in all_results: try: if (r.get("result") == "FAIL"): matched.append({ "name": r.get("name"), "description": r.get("description"), "severity": r.get("severity"), "tags": r.get("tags") or [], "reason": r.get("reason"), "category": r.get("category"), }) except Exception: # Ignore malformed entries continue findings = matched except Exception as exc: try: app_logger.error("Rule engine error", extra={"error": str(exc)}) except Exception: pass findings = [] snippet = _make_snippet(js_text, max_chars=settings.ui.snippet_preview_len) return jsonify({ "ok": True, "final_url": outcome.final_url, "status_code": outcome.status_code, "bytes": outcome.bytes_fetched, "truncated": outcome.truncated, "sha256": outcome.sha256_hex, "artifact_path": outcome.saved_path, "findings": findings, # only FAILed rules "snippet": snippet, "snippet_len": len(js_text) }) @api_bp.get("/artifacts//") def get_artifact_raw(run_uuid, filename): # prevent path traversal if "/" in filename or ".." in filename: abort(400) run_dir = _resolve_results_path(run_uuid) full_path = Path(run_dir) / filename # if file is not there, give a 404 if not os.path.isfile(full_path): abort(404) # else return file return send_file(full_path, as_attachment=False)