Free-threaded (no-GIL) use-after-free in Admin_create_topics: borrowed NewTopic from PyList_GET_ITEM dereferenced as a C struct against a stale count

Heads-up ahead of any free-threading support: `AdminClient.create_topics` (and the same-shaped `create_partitions` / topic-name paths) take a borrowed list item, cast it to a C struct, and dereference its fields while another thread can free it.

**This is latent today, not live.** The extension does not declare `Py_MOD_GIL_NOT_USED`, so a free-threaded interpreter **re-enables the GIL** on import and the code is safe as shipped. It becomes reachable only when the GIL is forced off (`PYTHON_GIL=0` / `-Xgil=0`), i.e. what will happen the day the module opts into free-threading. This is a "fix before advertising free-threading" report, not a "you are currently broken" one.

### Site

`src/confluent_kafka/src/Admin.c:609` on current `master`:

```c
tcnt = (int)PyList_Size(topics); // captured once (line 576)
c_objs = malloc(sizeof(*c_objs) * tcnt);
for (i = 0; i < tcnt; i++) {
NewTopic *newt = (NewTopic *)PyList_GET_ITEM(topics, i); // borrowed, no incref
r = PyObject_IsInstance((PyObject *)newt, ...); // reads newt->ob_type
c_objs[i] = rd_kafka_NewTopic_new(
newt->topic, newt->num_partitions, newt->replication_factor, ...); // 623-624
}
```

`newt` is borrowed, never increfed; it is cast to `(NewTopic *)` and its `char *topic` and two `int` fields are read straight out of the object. Two failure modes under `Py_GIL_DISABLED`, both driven by another thread mutating `topics`:

1. **stale `tcnt`** - the length is read once. If the list shrinks, `PyList_GET_ITEM(topics, i)` indexes past `ob_item` and the out-of-bounds slot is dereferenced as a `NewTopic`.
2. **freed element** - if element `i` is replaced, the old `NewTopic` drops to zero references and is freed; `newt` still points into freed memory and `newt->topic` is handed to librdkafka.

The whole parse-and-convert loop runs synchronously before `rd_kafka_CreateTopics` (line 680), so **no broker is required** - the `rd_kafka_t` handle is created at `AdminClient` construction and a bogus `bootstrap.servers` is enough.

The same borrow-then-dereference shape is at `Admin.c:864` (`create_partitions`, `NewPartitions *`) and the topic-name loops at 751 / 2513; I reproduced only `create_topics` and list the others as the same shape, not separately verified.

### Measured

confluent-kafka `2.15.0` built from source against librdkafka on a `python3.14.0rc1t` venv. 6 caller threads call `create_topics` on a bogus AdminClient while 3 threads mutate the list; 10 rounds per arm:

| arm | result |
|---|---|
| mutate, GIL off (`PYTHON_GIL=0`) | **10/10 SIGSEGV** |
| control - no mutator thread | clean 10/10 |
| control - mutate a different list `create_topics` never reads | clean 10/10 |
| control - same mutation, GIL left on | clean 10/10 |

The GIL-on control is the one that matters most: it is the current shipped behaviour, and it is clean - that is the whole "latent" point in one line.

Reproducer (ft_ckafka.py)

```python
"""confluent-kafka `Admin.c:609` — a borrowed NewTopic dereferenced as a C struct.

The exact shape, from confluent-kafka 2.15.0 `src/confluent_kafka/src/Admin.c`:

tcnt = (int)PyList_Size(topics); // 576, captured once
c_objs = malloc(sizeof(*c_objs) * tcnt); // 607
for (i = 0; i < tcnt; i++) {
NewTopic *newt = (NewTopic *)PyList_GET_ITEM(topics, i); // 609 — BORROWED
r = PyObject_IsInstance((PyObject *)newt, ...); // reads newt->ob_type
...
c_objs[i] = rd_kafka_NewTopic_new(
newt->topic, newt->num_partitions, newt->replication_factor, ...); // 623-624
// ^ char* and two ints read straight out of the C struct

`newt` is never increfed. Two ways this is a use-after-free under
`Py_GIL_DISABLED`, both exercised here by a second thread mutating `topics`
while `create_topics` runs:

(A) stale tcnt. tcnt is read once. If the list shrinks before the loop reaches
index i, PyList_GET_ITEM(topics, i) indexes past ob_item — an OOB read of a
freed slot — and the garbage is dereferenced as `newt->ob_type` /
`newt->topic`.

(B) freed element. If the list keeps its length but element i is replaced, the
old NewTopic loses its last reference and is freed. `newt` still points at
it, and `newt->topic` (a `char *` inside freed memory) is handed to
librdkafka.

None of this needs a broker. AdminClient creates its `rd_kafka_t` handle at
construction without connecting; the whole parse-and-convert loop runs
synchronously in the calling thread, before `rd_kafka_CreateTopics` is called at
line 680. A bogus bootstrap.servers is enough.

Run under the free-threaded build; a SIGSEGV / bus error / malloc abort is the
positive result. Compare against `ctrl_*` — same code, one property removed each.
"""
import os
import sys
import threading
import time

from confluent_kafka.admin import AdminClient, NewTopic

# a bogus broker: the handle is created, nothing connects
ADMIN_CONF = {"bootstrap.servers": "127.0.0.1:9", "socket.timeout.ms": 10}

N_CALLERS = int(os.environ.get("N_CALLERS", 6))
N_MUTATORS = int(os.environ.get("N_MUTATORS", 3))
LIST_LEN = int(os.environ.get("LIST_LEN", 64))
SECONDS = float(os.environ.get("SECONDS", 8))
# mutate=True -> the concurrent hazard (this file)
# mutate=False -> ctrl_A_no_mutator
MUTATE = os.environ.get("MUTATE", "1") == "1"
# decoy=True -> mutate a *different* list create_topics never reads
DECOY = os.environ.get("DECOY", "0") == "1"

def fresh_topics(n):
return [NewTopic(f"t{i}", num_partitions=1, replication_factor=1)
for i in range(n)]

def main():
gil = getattr(sys, "_is_gil_enabled", lambda: True)()
print(f"py={sys.version.split()[0]} gil={gil} "
f"callers={N_CALLERS} mutators={N_MUTATORS if MUTATE else 0} "
f"len={LIST_LEN} decoy={DECOY}", flush=True)

admin = AdminClient(ADMIN_CONF)
topics = fresh_topics(LIST_LEN)
decoy = fresh_topics(LIST_LEN)
stop = threading.Event()
start = threading.Barrier(N_CALLERS + (N_MUTATORS if MUTATE else 0) + 1)
calls = [0] * N_CALLERS

def caller(t):
start.wait()
n = 0
while not stop.is_set():
try:
admin.create_topics(topics, request_timeout=0.001)
except Exception:
pass # broker errors, validation errors — irrelevant
n += 1
calls[t] = n

def mutator():
start.wait()
target = decoy if DECOY else topics
i = 0
while not stop.is_set():
# (B) replace an element: old NewTopic drops to zero refs
target[i % len(target)] = NewTopic(
f"m{i}", num_partitions=1, replication_factor=1)
# (A) shrink then regrow: makes the captured tcnt stale mid-loop
if i % 7 == 0:
del target[:]
target.extend(fresh_topics(LIST_LEN))
i += 1

threads = [threading.Thread(target=caller, args=(t,))
for t in range(N_CALLERS)]
if MUTATE:
threads += [threading.Thread(target=mutator) for _ in range(N_MUTATORS)]

for th in threads:
th.start()
start.wait()
time.sleep(SECONDS)
stop.set()
for th in threads:
th.join()

print(f"clean: {sum(calls)} create_topics calls survived", flush=True)
return 0

if __name__ == "__main__":
raise SystemExit(main())
```

Run with `PYTHON_GIL=0`. The controls are env-var variants of the same script: `MUTATE=0` (no mutator), `DECOY=1` (mutate a different list), and running without `PYTHON_GIL=0` (GIL on).

### A possible fix

When free-threading is on the roadmap: incref `newt` for the body of the loop (or hold the list under a critical section), and re-read the list length rather than trusting the captured `tcnt`.

### Not claimed

No severity - reachable only with the GIL disabled, which is not the default for this module today. No exploitability - the observed faults are consistent with the mechanism; I did not build a primitive.