BOLA / IDOR Vulnerability Scanner

BOLA / IDOR Vulnerability Scanner

1. System Architecture & Prerequisites

• Python 3.10+ (stdlib only: re, os, json, argparse, urllib.request, urllib.error, pathlib)
• Target: local source tree (Express/Node or FastAPI route handlers) AND/OR a live localhost API
• No external dependencies; HTTP requests use urllib.request (no requests needed)

2. Input/Output Data Contracts

Input (CLI args):

{
  "type": "object",
  "properties": {
    "source": { "type": "string", "description": "Source root with route handlers (.js/.ts/.py)" },
    "base_url": { "type": "string", "description": "Live localhost API base, e.g. http://127.0.0.1:8000" },
    "user_id": { "type": "string", "description": "Authenticated user ID used for tampering tests" },
    "report_dir": { "type": "string", "description": "Output directory", "default": "./bola-reports" }
  },
  "required": ["source"]
}

Runtime tampering test contract (POST {base_url}/test):

{ "method": "GET", "url": "/api/orders/1", "path": "/api/orders/:id", "headers": {},
  "cookies": {}, "tampered_ids": [2, 3, 1001], "userId": "user-1" }

Output artifacts:

• {report_dir}/bola_report.json — structured BOLA findings
• {report_dir}/express_requireOwnership.js — Express middleware guard
• {report_dir}/fastapi_dependency.py — FastAPI dependency guard

3. Production Reference Implementation

#!/usr/bin/env python3
"""BOLA / IDOR Vulnerability Scanner — static route analysis + runtime tampering tests + guard generation."""

import re
import os
import sys
import json
import argparse
import urllib.request
import urllib.error
from pathlib import Path

# ─── Static Analysis ────────────────────────────────────────────────────────────

# Route handlers that fetch a resource by ID without an owner/tenant check.
ID_FETCH_PATTERNS = [
    re.compile(r"""(?:req\.params\.id|request\.params\.id|ctx\.params\.id|params\.id|path_params\[\s*["']id)"""),
    re.compile(r"""(?:getById|findById|find_by_id|get_object_or_404)""", re.IGNORECASE),
    re.compile(r"""(?:\.findOne\(|\.findOne\(|\.find\(|\.findById\(|Query\(.*\)\.get\()""", re.IGNORECASE),
    re.compile(r"""(?:\.get\(\s*["']?(?:resource|record|obj)\b|\.fetch\(\s*["']?[a-z_]*id)""", re.IGNORECASE),
]

OWNER_CHECK_PATTERNS = [
    re.compile(r"""(?:req\.user\.(?:id|sub|userId)|request\.user\.(?:id|sub)|current_user\.(?:id|sub))""", re.IGNORECASE),
    re.compile(r"""(?:owner_id|user_id|tenant_id|account_id|organization_id)""", re.IGNORECASE),
    re.compile(r"""(?:\.filter\(\s*.*(?:owner|user|tenant)|\.where\(\s*.*(?:owner|user|tenant))""", re.IGNORECASE),
    re.compile(r"""(?:authorize|is_owner|ownership|can_access|has_access|permission)""", re.IGNORECASE),
    re.compile(r"""(?:ensure_owner|require_owner|requireOwnership|belongs_to)""", re.IGNORECASE),
    re.compile(r"""(?:if\s*\(\s*[a-zA-Z_]+\.(?:owner|user)(?:Id|_id)?\s*(?:!==|!=)\s*(?:req\.user|request\.user|current_user))""", re.IGNORECASE),
]

ROUTE_MOUNT_PATTERNS = [
    re.compile(r"""(?:app\.(?:get|post|put|patch|delete)\(|router\.(?:get|post|put|patch|delete)\(|@(?:app|router)\.(?:get|post|put|patch|delete))"""),
    re.compile(r"""(?:(?:FastAPI|APIRouter)\.(?:get|post|put|patch|delete)\(|@(?:router|app)\.(?:get|post|put|patch|delete))"""),
]

SKIP_DIRS = {"node_modules", ".git", "__pycache__", "venv", ".venv", "dist", "build", ".next"}
SOURCE_EXTS = {".js", ".ts", ".py"}

HIGH_RISK_ROUTE_HINT = re.compile(r"""(?::id|{id}|<int:.*id|/me/|/accounts/|/orders/|/users/|/profiles/|/documents/|/transactions/)""", re.IGNORECASE)


def static_scan(source: Path) -> list:
    """Scan route handlers; return list of candidate IDOR findings."""
    findings = []
    for dirpath, dirnames, filenames in os.walk(source):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for fname in filenames:
            if Path(fname).suffix not in SOURCE_EXTS:
                continue
            fpath = Path(dirpath) / fname
            try:
                lines = fpath.read_text(encoding="utf-8", errors="replace").splitlines()
            except OSError:
                continue

            for i, line in enumerate(lines):
                if not any(p.search(line) for p in ROUTE_MOUNT_PATTERNS):
                    continue
                window_start = max(0, i - 2)
                window_end = min(len(lines), i + 25)
                context = "\n".join(lines[window_start:window_end])

                has_id_fetch = any(p.search(context) for p in ID_FETCH_PATTERNS)
                has_owner_check = any(p.search(context) for p in OWNER_CHECK_PATTERNS)
                is_high_risk = bool(HIGH_RISK_ROUTE_HINT.search(line))

                if has_id_fetch:
                    judgment = "VULNERABLE" if not has_owner_check else "review"
                    if "review" == judgment and not has_owner_check:
                        judgment = "VULNERABLE"
                    findings.append({
                        "file": str(fpath),
                        "line": i + 1,
                        "route": line.strip()[:160],
                        "snippet_low": context[:400],
                        "fetches_by_id": has_id_fetch,
                        "has_owner_check": has_owner_check,
                        "high_risk_path": is_high_risk,
                        "status": judgment,
                        "confidence": "high" if (has_id_fetch and not has_owner_check and is_high_risk) else "medium",
                        "cwe": "CWE-639",
                        "recommendation": "Verify authenticated user owns the requested object before returning it.",
                    })
    return findings


# ─── Runtime Tampering Tests ────────────────────────────────────────────────────

DEFAULT_TAMPER_IDS = [1, 2, 3, 1001, 99999]


def http_request(method: str, url: str, body: bytes = None, headers: dict = None):
    """Raw HTTP request via urllib. Returns (status, body_bytes)."""
    req = urllib.request.Request(url, data=body, method=method)
    req.add_header("Content-Type", "application/json")
    req.add_header("Accept", "application/json")
    for k, v in (headers or {}).items():
        req.add_header(k, v)
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            return resp.status, resp.read()
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read()
    except urllib.error.URLError as exc:
        raise RuntimeError(f"Network error reaching {url}: {exc.reason}")


def runtime_scan(base_url: str, user_id: str, static_paths: list) -> list:
    """Run tampering tests against live localhost API endpoints."""
    if not base_url:
        return []
    base_url = base_url.rstrip("/")
    results = []

    paths = set()
    for f in static_paths:
        m = re.search(r"""(?:['"`])((?:/[a-zA-Z0-9_\-{}:]+)+)""", f.get("route", ""))
        if m:
            paths.add(m.group(1))
    if not paths:
        paths = {"/api/orders/1", "/api/users/1", "/api/accounts/1"}

    for route in sorted(paths):
        resource_uri = route.split("?")[0]
        id_marker = re.search(r"(/)([^/]+)$", resource_uri)
        template_path = route
        for tam_id in DEFAULT_TAMPER_IDS:
            # Replace trailing segment (id or :id or {id}) with the tampered id.
            trial_uri = re.sub(r":[A-Za-z_][A-Za-z0-9_]*|\{[A-Za-z_][A-Za-z0-9_]*\}|[0-9]+$", str(tam_id), resource_uri, count=1)
            method = "POST"
            test_payload = {
                "method": "GET",
                "url": trial_uri,
                "path": template_path,
                "headers": {"Authorization": f"Bearer auth-{user_id}"},
                "cookies": {},
                "tampered_ids": DEFAULT_TAMPER_IDS,
                "userId": user_id,
            }
            test_endpoint = f"{base_url}/test"
            try:
                status, body = http_request(method, test_endpoint, json.dumps(test_payload).encode("utf-8"))
            except RuntimeError as exc:
                results.append({
                    "route": route, "tampered_id": tam_id, "test_endpoint": test_endpoint,
                    "status": "unreachable", "error": str(exc),
                })
                continue

            if status >= 200 and status < 300:
                verdict = "VULNERABLE" if status == 200 else "review"
                results.append({
                    "route": route,
                    "tampered_id": tam_id,
                    "requested_uri": trial_uri,
                    "response_status": status,
                    "response_preview": body[:200].decode("utf-8", errors="replace"),
                    "verdict": verdict,
                    "finding": "User can access foreign object by tampered ID" if status == 200 else
                               "Server responded 2xx on tampered ID — review access policy",
                })
            else:
                results.append({
                    "route": route, "tampered_id": tam_id, "requested_uri": trial_uri,
                    "response_status": status, "verdict": "PROTECTED",
                    "finding": "Request rejected (expected for protected object)",
                })
    return results


# ─── Guard Generation ───────────────────────────────────────────────────────────

def generate_express_guard(report_dir: Path) -> Path:
    code = """const { ObjectId } = require('mongodb'); // or omit for SQL/ORM

// requireOwnership(resourceLoader, options)
// Express middleware that guarantees the authenticated user OWNS the requested
// object before the handler runs. Prevents BOLA/IDOR (CWE-639).
//
// usage:
//   const orderRepo = { byId: (id) => db.orders.findOne({ _id: id }) };
//   app.get('/api/orders/:id', requireOwnership(
//     async (req) => orderRepo.byId(req.params.id),
//     { ownerField: 'userId' }
//   ), (req, res) => { res.json(req.resource); });

function requireOwnership(resourceLoader, options = {}) {
  const { ownerField = 'userId', idField = 'id', onDeny } = options;

  return async function requireOwnershipMiddleware(req, res, next) {
    if (!req.user || !req.user.id) {
      return res.status(401).json({ error: 'Unauthorized' });
    }

    const rawId = req.params[idField];
    if (rawId === undefined) {
      return res.status(400).json({ error: 'Missing resource id' });
    }

    let resource;
    try {
      resource = await resourceLoader(req, rawId, res.locals);
    } catch (err) {
      return res.status(404).json({ error: 'Resource not found' });
    }

    if (!resource) {
      return res.status(404).json({ error: 'Resource not found' });
    }

    const ownerId = resource[ownerField];
    const requesterId = String(req.user.id);

    if (String(ownerId) !== requesterId) {
      if (typeof 'function') {
        return onDeny(req, res, next);
      }
      return res.status(403).json({ error: 'Forbidden: you do not own this resource' });
    }

    // Optionally narrow the returned object to non-sensitive fields.
    req.resource = resource;
    next();
  };
}

module.exports = { requireOwnership };
"""
    out = report_dir / "express_requireOwnership.js"
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(code, encoding="utf-8")
    return out


def generate_fastapi_guard(report_dir: Path) -> Path:
    code = '''"""FastAPI tenant-isolated authorization dependency.

usage:
    @router.get("/api/orders/{order_id}")
    async def get_order(
        order_id: int,
        session: Session = Depends(get_session),
        order: Order = Depends(require_ownership(Order, "owner_id")),
    ):
        return order
"""
from typing import Type, Callable
from fastapi import Depends, HTTPException, status
from sqlalchemy import select


def require_ownership(
    model: Type,
    owner_field: str = "owner_id",
    id_field: str = "id",
    id_type: type = int,
):
    """Return a FastAPI dependency enforcing owner-of-object authorization."""

    async def dependency(
        resource_id: id_type,  # type: ignore[valid-type]  # path param name
        request,
        db_session,
    ):
        # The current authenticated user must be resolvable from request.state.
        user = getattr(request.state, "user", None)
        if user is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Unauthorized",
            )

        stmt = select(model).where(model.__table__.c[id_field] == resource_id)
        resource = (await db_session.execute(stmt)).scalar_one_or_none()

        if resource is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Resource not found",
            )

        owner_value = getattr(resource, owner_field)
        if owner_value is None or str(owner_value) != str(getattr(user, "id", None)):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Forbidden: you do not own this resource",
            )

        return resource

    return dependency
'''
    out = report_dir / "fastapi_dependency.py"
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(code, encoding="utf-8")
    return out


# ─── Orchestration ──────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(
        description="BOLA / IDOR Vulnerability Scanner — static + runtime authorization checks.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--source", required=True, help="Source root with route handlers")
    parser.add_argument("--base-url", default=None, help="Live localhost API base, e.g. http://127.0.0.1:8000")
    parser.add_argument("--user-id", default="testuser-1", help="Authenticated user ID for tampering tests")
    parser.add_argument("--report-dir", default="./bola-reports", help="Output directory")
    args = parser.parse_args()

    source = Path(args.source).resolve()
    if not source.is_dir():
        print(f"[ERROR] Source directory does not exist: {source}")
        raise SystemExit(1)

    report_dir = Path(args.report_dir).resolve()
    report_dir.mkdir(parents=True, exist_ok=True)

    print(f"[INFO] Static scanning: {source}")
    static_findings = static_scan(source)
    print(f"[INFO] {len(static_findings)} candidate IDOR route(s) found")
    for f in static_findings:
        print(f"  [{f['status']}] {f['file']}:{f['line']} {f['route'][:80]}")

    runtime_findings = []
    if args.base_url:
        print(f"[INFO] Runtime tampering tests against {args.base_url} (user={args.user_id})")
        runtime_findings = runtime_scan(args.base_url, args.user_id, static_findings)
        for r in runtime_findings:
            print(f"  [{r.get('verdict', r.get('status'))}] {r.get('route')} tamper id={r.get('tampered_id')} -> HTTP {r.get('response_status', 'N/A')}")

    express_guard = generate_express_guard(report_dir)
    fastapi_guard = generate_fastapi_guard(report_dir)

    report = {
        "scanned_source": str(source),
        "live_api": args.base_url,
        "static_findings": static_findings,
        "runtime_findings": runtime_findings,
        "generated_guards": {
            "express": str(express_guard),
            "fastapi": str(fastapi_guard),
        },
        "summary": {
            "static_vulnerable_count": sum(1 for f in static_findings if f["status"] == "VULNERABLE"),
            "runtime_vulnerable_count": sum(1 for r in runtime_findings if r.get("verdict") == "VULNERABLE"),
            "runtime_protected_count": sum(1 for r in runtime_findings if r.get("verdict") == "PROTECTED"),
        },
        "remediation": "Mount requireOwnership (Express) or require_ownership (FastAPI) on every object-level route.",
    }

    report_path = report_dir / "bola_report.json"
    report_path.write_text(json.dumps(report, indent=2, default=str), encoding="utf-8")
    print(f"\n[INFO] Report:         {report_path}")
    print(f"[INFO] Express guard:  {express_guard}")
    print(f"[INFO] FastAPI guard:  {fastapi_guard}")
    print(f"[INFO] Summary: {report['summary']}")


if __name__ == "__main__":
    main()

4. Execution Protocol & Step-by-Step Workflow

1. Point the scanner at your local route-handler source:

   python bola-idor-vulnerability-scanner.md --source ./api --report-dir ./bola-reports
   

2. Review static findings: each row shows file/line, whether the handler fetches by ID, and whether an owner check is present.
3. If an API is running locally, enable live tampering tests:

   python bola-idor-vulnerability-scanner.md --source ./api --base-url http://127.0.0.1:8000 --user-id user-1
   

4. Confirm your API exposes the test harness POST /test described in the data contract, returning proxied responses.
5. Inspect bola_report.json for static_findings and runtime_findings + verdicts.
6. Deploy express_requireOwnership.js (Express) or fastapi_dependency.py (FastAPI) guards onto vulnerable routes.
7. Re-run the tamper tests; previously VULNERABLE routes must now return 403/404.

5. Edge Cases & Error Handling

• Unreachable live API → tampering tests record status: "unreachable" with the underlying network error instead of crashing.
• Non-JSON responses from POST /test are tolerated; preview is truncated to 200 bytes.
• Static scanner uses a 25-line context window to detect owner checks; complex authorization (RBAC lookups, policy DSLs) may yield false positives flagged confidence: medium.
• Routes without a trailing :id/{id} segment fall back to default probe paths; the runtime probe always rewrites the final URI segment to the tampered ID.
• Generated guard code is written to disk even when no live API is provided, so remediation assets are always available.
• Rollback: generated guards are new files; removing them fully reverts the codebase. No existing file is modified by this skill.