Python backdoor detection: inventory eval, sockets, pickle

A Python backdoor is usually pickle, eval, or a subprocess you did not mean to expose, not a forty-line magic trick.

If your app deserializes bytes from a user, a queue, or a cache, those bytes can become an object graph that runs code. The official pickle docs say this in the first paragraph. Tutorials still treat pickle as a convenient save format.

The usual mistake is signing the blob and calling it safe. A signature only helps if the attacker does not have the key, and it does nothing if you also accept an unsigned format on another path.

This page is why those forty-line demos work, which sinks to refuse, and the tests that prove an uploaded payload cannot become a shell.

The 3.13.15 pickle page, last updated 22 August 2026, still opens with: the pickle module is not secure. Only unpickle data you trust. Bandit still ships B301 for that loader, B307 for eval, and B602 for a subprocess call with shell=True. This page is how a Python team finds those shapes and shuts them.

Keep the Python hatch list for the four everyday misses. Read injection when the interpreter is SQL or the shell, Flask locks when the app is Pallets, and the secure coding checklist for the rest of the request surface. This page is detection, inventory, and privilege. Not construction.

The three hatches Bandit already names

CWE-94 is code injection. CWE-502 is a loader that runs during deserialize. CWE-78 is a shell string. The shared shape is the same: a process that was supposed to treat bytes as data starts treating them as grammar. In CPython that almost always shows up as one of three calls.

current 3.13 pickle page the same day. The next sentence after the banner says a crafted stream can run during load. The page then names JSON as the safer format. That is the whole control. Do not keep a find_class allowlist and hope the next vendor blob stays inside it.

eval and exec are the other two. Bandit B307 says consider ast.literal_eval. That helper parses Python literals. It does not run names, calls, or attribute chains. If the string is a config blob, JSON is still the better parser. If the string is a formula you invented, write a grammar. Do not call eval on a request body and then argue the regex in front of it.

import ast
import json

def load_cache(path: Path) -> dict:
 raw = path.read_text(encoding="utf-8")
 data = json.loads(raw)
 if not isinstance(data, dict):
 raise ValueError("cache must be an object")
 return data

def parse_literal(text: str):
 return ast.literal_eval(text)

load_cache and parse_literal are the named pair. A Redis string, a celery result, or a file that used to be a pickle becomes text you can print in a ticket. The miss is the one-liner people still paste:

# BAD: do not ship
# obj = pickle.loads(blob)
# result = eval(request_text)

An unexpected socket.connect is the third shape. A worker that only talks to Postgres on 5432 has no reason to open a new outbound socket to an address that is not in the runbook. The 3.13 audit table lists socket.connect, socket.bind, http.client.connect, and urllib.Request. Those events are how you see the call without reading every vendor file by hand.

Inventory the tree before you trust a worker

Detection starts as a list of files, not a story. inventory_tree is the named scan. It walks the app package and reports the four call shapes. Put it in CI. Fail the job on a new hit that is not in the allow file.

from pathlib import Path

HATCH = (
 "eval(",
 "exec(",
 "compile(",
 "pickle.loads",
 "pickle.load(",
 "dill.loads",
 "socket.connect",
 "socket.create_connection",
 "shell=True",
 "os.system",
)

SKIP = {".venv", "venv", "__pycache__", "node_modules"}

def inventory_tree(root: Path) -> list[str]:
 hits: list[str] = []
 for path in root.rglob("*.py"):
 if any(part in SKIP for part in path.parts):
 continue
 text = path.read_text(encoding="utf-8", errors="replace")
 for i, line in enumerate(text.splitlines(), start=1):
 if any(token in line for token in HATCH):
 hits.append(f"{path}:{i}:{line.strip()}")
 return hits

Run it against the repo root. A hit is a review, not a merge. The sibling grep if you do not want a Python helper:

rg -n "\\beval\\(|\\bexec\\(|pickle\\.loads|pickle\\.load\\(|dill\\.loads|socket\\.connect|socket\\.create_connection|shell\\s*=\\s*True|os\\.system|os\\.popen" \
 --glob '!venv' --glob '!.venv' --glob '!__pycache__'

Bandit is the AST pass. If CI already runs it, keep B301, B307, B403, and B602 on. A # nosec on those four needs a ticket id and a name for the replacement. A style-only pre-commit job will not see a new pickle.loads in a worker you added last Tuesday.

Inventory the image too. A wheel you did not pin can grow a sitecustomize.py or a .pth file that imports on startup. List site.getsitepackages(), then grep those directories for the same tokens. A lockfile that only pins hashes is not a substitute for that scan. It is the other half.

TREE inventory_tree(root)
 eval / exec / pickle / socket.connect
 a new hit fails CI

HOOK deny_runtime(event, args)
 evaluator events raise
 socket.connect only if host:port is listed

DROP no extra capability
 outbound only to the listed pair
 uid is not 0

Deny exec, pickle, and stray sockets at runtime

PEP 578 shipped audit events in 3.8. The 3.13 audit table on 22 August 2026 still lists the evaluator events, plus socket.connect and subprocess.Popen. sys.addaudithook is how you subscribe. A hook that raises aborts the call.

deny_runtime is the named hook. Install it before the app imports workers. A hook added after pickle has already loaded a blob is late.

import sys

DENY = frozenset({"exec", "compile", "pickle.find_class"})
ALLOWED_NET = frozenset({
 ("db.internal.example", 5432),
})

def deny_runtime(event, args):
 if event in DENY:
 raise RuntimeError("blocked " + event)
 if event == "socket.connect" and len(args) > 1:
 address = args[1]
 if address not in ALLOWED_NET:
 raise RuntimeError("blocked socket.connect")

def install_deny() -> None:
 sys.addaudithook(deny_runtime)

Call install_deny from sitecustomize.py or from the first line of the process entry. ALLOWED_NET is a pair you wrote down. A cache host, a queue host, an identity provider. If the worker has no outbound need, leave the set empty and every socket.connect fails. That is the conservative reading.

Do not log the full args tuple for exec into a ticket that leaves your host. Log the event name and the file from sys._getframe only if you already accept that event as noisy. Prefer failing closed.

http.client.connect and urllib.Request are sibling events. If the app uses httpx or requests, those libraries still open a socket. The socket.connect deny covers the TCP open. Keep the host allowlist in one place. Do not copy it into three helpers.

Least privilege after the hook

A hook that raises is still running as whatever uid started the process. If that uid can write /etc or open raw sockets, a later miss still matters. Drop the extra rights after install_deny and after the process has opened the sockets it was written to use.

On a systemd unit the flags that match this page are User= not root, NoNewPrivileges=yes, ProtectSystem=strict, ProtectHome=yes, and a RestrictAddressFamilies= list that only names AF_INET and AF_UNIX if you need those. If the worker never leaves the host, PrivateNetwork=yes is the stronger lock. exec man page for those names.

Linux capabilities are the other knob. A web worker does not need CAP_NET_RAW, CAP_SYS_ADMIN, or CAP_SYS_PTRACE. Drop them in the unit or in the container securityContext. A Kubernetes pod that mounts the Docker socket is not least privilege. Neither is privileged: true.

Network policy is the third. If ALLOWED_NET names one host, the namespace NetworkPolicy should name that same host. The hook and the policy should agree. When they disagree, the policy is the one the kernel will still enforce after a hook is removed.

# systemd fragment, names only
# User=app
# NoNewPrivileges=yes
# ProtectSystem=strict
# RestrictAddressFamilies=AF_INET AF_UNIX
# AmbientCapabilities=
# CapabilityBoundingSet=

Do not run the worker as root to bind 443. Bind 8080 and put a proxy in front. This page stops at the Python process.

Prove the inventory and the hook

You are not walking an implant. You are proving inventory_tree is empty except for the lines you listed, and that deny_runtime raises on the three events.

python -c "from pathlib import Path; from inventory import inventory_tree; print(inventory_tree(Path('.')))"
# Expect: [] or only paths in the signed allow file

Then prove the hook. Import install_deny first. Call a tiny probe that would have been a hatch. Expect RuntimeError.

from deny import install_deny

install_deny()

try:
 compile("1 + 1", "<probe>", "eval")
except RuntimeError as err:
 print(err)
# Expect: blocked compile

The probe uses compile on a constant. It does not load a stream, open a socket, or evaluate request text. If that line does not raise, install_deny was not first. A 200 from your app after that miss is not a pass.

A third check is the listen and connect set at runtime. After the process is up, read /proc/self/net/tcp or ss -tpn and confirm the remote addresses match ALLOWED_NET. A new outbound 443 to a host you did not list is a review, even if the hook failed to see the library that opened it.