How to Load Untrusted YAML Safely in Python and Stop Code Execution, Alias Bombs, Duplicate Keys, and the Norway Problem
Watch yaml.load run a function from a file, see why safe_load is only the first fix, then build a strict PyYAML loader that refuses alias bombs, duplicate keys and the Norway problem, with 49 tests...
A YAML file looks like harmless data: a few names, a few numbers, a list. Then someone loads it with the wrong function and discovers that the file can also be a program. YAML sits behind CI pipelines, Kubernetes manifests and countless application settings, and it often comes from people you do not control: a plugin author, a pull request, a template someone downloaded. In Python the library that reads it is usually PyYAML, and its documentation is blunt: “yaml.load is as powerful as pickle.load and so may call any Python function.”
Table Of Content
- The words you need first
- What you will build, and what it does not do
- Prerequisites
- Step 1: Set up the lab and check your versions
- Step 2: Watch a YAML file call a Python function
- Step 3: Find out which loaders are dangerous
- Step 4: Switch to safe_load and see what it refuses
- Step 5: Meet aliases, and see why a small file can describe a huge one
- Step 6: Catch the key that appears twice
- Step 7: See what safe_load guesses when you do not use quotes
- Step 8: Build a loader that only returns plain data
- 8a. Limits and one error type
- 8b. Check the whole document before building any of it
- 8c. The loader class
- 8d. Replace the type guesses
- 8e. The entry points
- 8f. Try it on every nasty document
- 8g. Compare three loaders side by side
- Step 9: Check the shape of the data with a schema
- Step 10: Test every promise, then break the loader on purpose
- Step 11: Stop the next unsafe call before it merges
- Step 12: Confirm the whole thing works end to end
- Common mistakes and gotchas
- Where to go next
In this tutorial you will see that for yourself with a harmless test, learn why yaml.safe_load is the right first fix, and then learn what it still lets through: a 1,914-byte file that describes more data than any computer can hold, a key that appears twice so that the last one silently wins, and plain words such as NO that quietly turn into booleans. You will finish with a small loader, a schema check, 49 tests and a CI gate. This is the YAML chapter of a series on loading data you do not trust, after our tutorials on pickle, XML and decompression bombs. It also pairs with our tutorial on kubectl’s KYAML output and YAML’s Norway bug, which explains the same type-guessing trap from the Kubernetes side.
The words you need first
A loader is the PyYAML class that turns YAML text into Python objects; yaml.safe_load(text) is the same as loading with yaml.SafeLoader. A tag is a label on a value that says what type it is, such as !!str. PyYAML also understands tags that name Python things, for example !!python/object/apply:os.mkdir, and that is where the danger starts. An anchor (&name) marks a value and an alias (*name) reuses it, so you can write something once and refer to it many times. Implicit typing means that when no tag is written, the loader guesses the type from how the text looks: 42 becomes an integer and true becomes a boolean. Untrusted means any file you did not write and review yourself.
Loading happens in stages, and each problem in this tutorial lives in a different one:
YAML text --parser--> events --composer--> nodes --constructor--> objects
^
resolver gives every node a type tag- The constructor decides which tags may build which Python objects. Code execution lives here (Steps 2 to 4).
- The composer assembles the nodes and resolves aliases, which is where a small file can describe a huge one (Step 5).
- The resolver guesses the type of plain text, which is where
NObecomesFalse(Step 7).
Duplicate keys (Step 6) slip through at the moment the constructor fills a dictionary. Keep the stages in mind, because each fix later plugs into the stage where its problem occurs.
What you will build, and what it does not do
By the end you will have:
- measured, on PyYAML 6.0.3, which of its five loaders run code that a file asks for;
safeyaml.py, about 110 lines: a loader that only ever returns plain data, refuses Python tags, alias bombs, recursive aliases, duplicate keys, oversized files and bad encodings, and stopsNO,0755and2026-11-02from changing type;manifest.py, a short Pydantic schema that checks the shape of the data and rejects unknown keys;- 49 pytest tests and ten deliberate bugs that those tests catch;
yamlgate.py, a CI check that fails the build when anyone calls an object-building loader.
What this does not do: it does not make a file’s contents safe to act on. If your program reads a command name from a YAML file and runs it, that is your code’s decision, and you still need an allow-list. It covers PyYAML 6.0.3 only; other YAML libraries and other languages make different choices, so test yours the same way. And if you write every file yourself, safe_load is plenty.
Prerequisites
- Python 3.10 or newer. Every command below ran on Python 3.13.14 on Windows 11. I read
Requires-Pythonin each pinned package’s metadata: Pydantic 2.14.2 and pytest 9.2.0 need 3.10 or newer, ruamel.yaml 0.19.1 needs 3.9 or newer and PyYAML 6.0.3 needs 3.8 or newer. - A bash shell. On Windows I used Git Bash. Linux and macOS need nothing extra.
- Packages: PyYAML 6.0.3, Pydantic 2.14.2, pytest 9.2.0, Ruff 0.17.3 and ruamel.yaml 0.19.1, pinned in Step 1.
- Background: you can read a Python function, a dictionary and a list. No Docker, no cloud account and no real credentials are needed. The test documents are harmless: the only side effect any of them has is to create an empty folder inside a temporary folder.
Step 1: Set up the lab and check your versions
Make a folder and a virtual environment, which is a private set of installed packages for one project, so that nothing here touches your system Python. Then install the pinned packages:
mkdir yaml-lab && cd yaml-lab
python -m venv .venv
source .venv/bin/activate
pip install pyyaml==6.0.3 pydantic==2.14.2 pytest==9.2.0 ruff==0.17.3 ruamel.yaml==0.19.1On Windows in Git Bash the activation line is source .venv/Scripts/activate. Pinning means asking for exact versions, so that your results match the ones printed below. Every file in this tutorial goes in this one folder, and the first line of each code block is a comment with its file name.
Start with a tiny script that reports what you are running:
# step01_versions.py - what are we running?
import platform
import yaml
print("Python :", platform.python_version())
print("PyYAML :", yaml.__version__)
print("libyaml :", yaml.__with_libyaml__)
names = ("BaseLoader", "SafeLoader", "FullLoader", "Loader", "UnsafeLoader")
print("Loaders :", ", ".join(name for name in names if hasattr(yaml, name)))Run it with python step01_versions.py. Expected output:
Python : 3.13.14
PyYAML : 6.0.3
libyaml : True
Loaders : BaseLoader, SafeLoader, FullLoader, Loader, UnsafeLoaderTwo lines deserve a comment. libyaml: True means this PyYAML build includes the C parser called libyaml, which a faster loader named CSafeLoader uses; Step 8 measures it. The Loaders line lists the five loader classes you are about to compare. If your PyYAML version differs, later tables might differ too, and the PyYAML changelog will tell you why.
Step 2: Watch a YAML file call a Python function
First a helper file that every later step shares. scratch() makes a throw-away folder, moves into it and deletes it afterwards, so whatever a test document does, it can only do inside that folder. PAYLOADS is a set of test documents that Step 3 explains, and bomb() builds the alias bomb of Step 5. Ignore those two for now.
# labkit.py - small helpers shared by the step scripts
import contextlib
import os
import shutil
import tempfile
# Documents used by several steps. The only side effect any of them can have is
# creating a folder named CANARY_RAN inside the scratch folder.
PAYLOADS = {
"plain mapping": "{name: demo, size: 3}",
"!!set": "!!set {a}",
"!!python/tuple": "!!python/tuple [1, 2]",
"!!python/module": "!!python/module:os",
"!!python/name": "!!python/name:os.mkdir",
"!!python/object/new": '!!python/object/new:os.mkdir ["CANARY_RAN"]',
"!!python/object/apply": '!!python/object/apply:os.mkdir ["CANARY_RAN"]',
}
def bomb(levels, fan=9):
"""YAML where every level lists the previous level `fan` times, by alias."""
lines = ["l0: &l0 [" + ", ".join(['"lol"'] * fan) + "]"]
for n in range(1, levels + 1):
lines.append(f"l{n}: &l{n} [" + ", ".join([f"*l{n - 1}"] * fan) + "]")
return "\n".join(lines) + "\n"
@contextlib.contextmanager
def scratch():
"""Run inside a throw-away folder, then go back and delete it."""
old = os.getcwd()
folder = tempfile.mkdtemp(prefix="yamllab_")
os.chdir(folder)
try:
yield folder
finally:
os.chdir(old)
shutil.rmtree(folder, ignore_errors=True)The only side effect any test document has is to create a folder named CANARY_RAN. A canary in a coal mine is the classic early warning, and this folder plays the same role: if it exists afterwards, the loader ran code from the file.
Now the first experiment. The document is a single line of YAML, and the loader is UnsafeLoader, which PyYAML names honestly:
# step02_danger.py - a YAML file that makes the loader call a Python function
import os
import yaml
from labkit import scratch
doc = '!!python/object/apply:os.mkdir ["CANARY_RAN"]\n'
print("YAML text:", doc.strip())
with scratch():
print("folder exists before load:", os.path.isdir("CANARY_RAN"))
result = yaml.load(doc, Loader=yaml.UnsafeLoader)
print("load returned :", result)
print("folder exists after load :", os.path.isdir("CANARY_RAN"))Run python step02_danger.py:
YAML text: !!python/object/apply:os.mkdir ["CANARY_RAN"]
folder exists before load: False
load returned : None
folder exists after load : TrueThe file was not data; it was an instruction. The tag !!python/object/apply:os.mkdir means “import os, find mkdir and call it”, and the list ["CANARY_RAN"] after it is the argument list. The loader obeyed, the folder appeared, and load returned None because os.mkdir returns nothing. os.mkdir is harmless, but the same syntax can name any function that Python can import, including functions that start other programs. That is exactly what PyYAML’s warning about pickle.load means, and it is the weakness that CWE-502 describes as a product that “deserializes untrusted data without sufficiently ensuring that the resulting data will be valid.”
A common mistake is copying an old answer that says yaml.load(text, Loader=yaml.Loader). The name Loader is the dangerous one, as the next step shows.
Step 3: Find out which loaders are dangerous
PyYAML has five loaders, and not all of them are safe. This script feeds seven documents to each loader and reports one word per cell. RAN CODE means the canary folder appeared, refused means the loader raised an error, OBJECT means it built something other than plain data, and the other cells name the plain type that came back. Notice that os is imported at the top: FullLoader only looks up names in modules that the program has already imported.
# step03_matrix.py - which loaders turn Python tags into Python objects?
import os
import yaml
from labkit import PAYLOADS, scratch
COLUMNS = [("Base", "BaseLoader"), ("Safe", "SafeLoader"), ("Full", "FullLoader"),
("Loader", "Loader"), ("Unsafe", "UnsafeLoader")]
def attempt(loader_name, text):
"""Load `text`, then report what happened in one or two words."""
with scratch():
try:
value = yaml.load(text, Loader=getattr(yaml, loader_name))
except yaml.YAMLError:
return "refused"
if os.path.isdir("CANARY_RAN"):
return "RAN CODE"
if isinstance(value, (dict, list, set, tuple, str)):
return type(value).__name__
return "OBJECT"
print(f"{'document':<23}" + "".join(f"{label:<9}" for label, _ in COLUMNS))
for name, text in PAYLOADS.items():
cells = [attempt(loader, text) for _, loader in COLUMNS]
print(f"{name:<23}" + "".join(f"{cell:<9}" for cell in cells))
print()
print("FullLoader, !!python/name:os.mkdir ->",
yaml.load(PAYLOADS["!!python/name"], Loader=yaml.FullLoader))Run python step03_matrix.py:
document Base Safe Full Loader Unsafe
plain mapping dict dict dict dict dict
!!set dict set set set set
!!python/tuple list refused tuple tuple tuple
!!python/module str refused refused OBJECT OBJECT
!!python/name str refused OBJECT OBJECT OBJECT
!!python/object/new list refused refused RAN CODE RAN CODE
!!python/object/apply list refused refused RAN CODE RAN CODE
FullLoader, !!python/name:os.mkdir -> <built-in function mkdir>Read the columns from left to right. BaseLoader built nothing but dictionaries, lists and strings. SafeLoader accepted plain data and !!set and refused every Python tag, even the harmless !!python/tuple. FullLoader refused the two tags that call things (new and apply) and the module tag, but it built a tuple and handed back the function os.mkdir itself, as the last line shows. A function is not a call, yet it is far from plain data. Loader and UnsafeLoader behave identically here: both ran the code for new and for apply, and both gave back a module and a function object. The old name Loader is simply the unsafe loader.
PyYAML’s changelog explains how it got this way:
- 5.1 (March 2019) deprecated
yaml.loadand addedFullLoaderandUnsafeLoader(pull request 257). - 5.3.1 (March 2020) prevented arbitrary code execution during the
python/object/newconstructor (pull request 386). The National Vulnerability Database describes CVE-2020-1747 as arbitrary code execution “through the full_load method or with the FullLoader loader” in versions before 5.3.1. - 5.4 (January 2021) fixed CVE-2020-14343 by moving the arbitrary Python tags to
UnsafeLoader(pull request 472); NVD uses the same wording for versions before 5.4. - 6.0 (October 2021) made the
Loaderargument mandatory foryaml.load()(pull request 561).
The lesson is simple: of the loaders that return typed data, only SafeLoader refuses every Python tag, and yaml.safe_load is the function that uses it.
Step 4: Switch to safe_load and see what it refuses
yaml.safe_load(text) is the one-call form of loading with SafeLoader. This script runs all seven documents through it, then makes the mistake that PyYAML 6 turns into an error: calling yaml.load without saying which loader to use.
# step04_safe.py - the same documents through yaml.safe_load
import os
import textwrap
import yaml
from labkit import PAYLOADS, scratch
with scratch():
for name, text in PAYLOADS.items():
try:
line = f"{name:<23}loaded {yaml.safe_load(text)!r}"
except yaml.YAMLError as exc:
line = f"{name:<23}{type(exc).__name__}: {exc.problem}"
print(textwrap.fill(line, width=76, subsequent_indent=" " * 4))
print("folder created:", os.path.isdir("CANARY_RAN"))
try:
yaml.load("a: 1")
except TypeError as exc:
print("yaml.load(text) ->", exc)Run python step04_safe.py:
plain mapping loaded {'name': 'demo', 'size': 3}
!!set loaded {'a'}
!!python/tuple ConstructorError: could not determine a constructor
for the tag 'tag:yaml.org,2002:python/tuple'
!!python/module ConstructorError: could not determine a constructor
for the tag 'tag:yaml.org,2002:python/module:os'
!!python/name ConstructorError: could not determine a constructor
for the tag 'tag:yaml.org,2002:python/name:os.mkdir'
!!python/object/new ConstructorError: could not determine a constructor
for the tag 'tag:yaml.org,2002:python/object/new:os.mkdir'
!!python/object/apply ConstructorError: could not determine a constructor
for the tag 'tag:yaml.org,2002:python/object/apply:os.mkdir'
folder created: False
yaml.load(text) -> load() missing 1 required positional argument: 'Loader'Plain data and !!set load. Every Python tag raises ConstructorError with the message “could not determine a constructor for the tag”, which is how a safe loader says that it has no recipe for building such a thing. The line that matters most is folder created: False: nothing ran. The final line shows the PyYAML 6 rule from the changelog, that yaml.load needs a Loader argument, so old code that left it out now fails loudly instead of quietly running unsafe code.
How to verify this step: search your own code with grep -rn "yaml.load" src/ and read each hit. Step 11 replaces that manual search with an automatic gate.
This is where many tutorials stop, and for files you wrote yourself it is a fine place to stop. For files someone else wrote, safe_load has gaps. The next three steps show them with measurements, and Step 8 closes them.
Step 5: Meet aliases, and see why a small file can describe a huge one
An anchor names a value and an alias points back to it. They exist so that you can write a default once and reuse it. This script starts with the idea in its smallest form, a bomb with three levels and a fan-out of three, and proves that safe_load does not copy: the three entries of l1 are one and the same list. It then times larger bombs, a node that contains itself, and deep nesting. Each part prints under its own heading.
# step05_aliases.py - aliases let a small file describe a huge structure
import json
import time
import yaml
from labkit import bomb
print("=== a tiny bomb, three levels, fan-out 3")
print(bomb(2, fan=3), end="")
small = yaml.safe_load(bomb(2, fan=3))
print("l1[0] is l1[1]:", small["l1"][0] is small["l1"][1])
print("=== loading is quick, using the result is not")
print("levels bytes leaves load ms json.dumps ms json MB")
for levels in (3, 4, 5, 6):
text = bomb(levels)
start = time.perf_counter()
data = yaml.safe_load(text)
load_ms = (time.perf_counter() - start) * 1000
start = time.perf_counter()
out = json.dumps(data[f"l{levels}"])
dump_ms = (time.perf_counter() - start) * 1000
print(f"{levels:>6} {len(text):>7,} {9 ** (levels + 1):>12,} {load_ms:>8.2f}"
f" {dump_ms:>13.1f} {len(out) / 1e6:>8.1f}")
print("=== thirty levels")
text = bomb(30)
start = time.perf_counter()
data = yaml.safe_load(text)
load_ms = (time.perf_counter() - start) * 1000
print(f"{len(text):,} bytes loaded in {load_ms:.2f} ms")
print(f"leaves if you walk it: {9 ** 31:.2e}")
print("=== a node that contains itself")
loop = yaml.safe_load("&a [*a]")
print("loop[0] is loop:", loop[0] is loop)
try:
json.dumps(loop)
except ValueError as exc:
print("json.dumps ->", exc)
print("=== deep nesting without aliases")
for depth in (100, 5000):
try:
yaml.safe_load("[" * depth + "]" * depth)
print(f"depth {depth:>5}: loaded")
except RecursionError:
print(f"depth {depth:>5}: RecursionError")Run python step05_aliases.py and look at the first block:
l0: &l0 ["lol", "lol", "lol"]
l1: &l1 [*l0, *l0, *l0]
l2: &l2 [*l1, *l1, *l1]
l1[0] is l1[1]: TrueEach level lists the previous level three times by alias. That is the whole trick, known from XML as the “billion laughs” pattern: every level multiplies the size of the data you would get if you wrote it all out, while the file grows by one short line. bomb() in labkit.py does the same with a fan-out of nine. Now the measurements:
levels bytes leaves load ms json.dumps ms json MB
3 234 6,561 0.44 0.2 0.0
4 288 59,049 0.51 1.9 0.4
5 342 531,441 0.59 17.4 3.9
6 396 4,782,969 0.67 152.9 34.7Loading is quick because PyYAML’s composer does not copy. When it meets an alias it returns the node that carries the anchor (return self.anchors[anchor] in composer.py), so nine references cost nine pointers. Using the result is another matter. json.dumps has to write every leaf, and each extra level multiplies that work by nine: level 6 produced 34.7 MB of JSON, and by my arithmetic level 8 would produce about 2.8 GB. Anything else that walks the structure pays the same price: logging it, templating it, copying it into another format, validating every element. Now the extreme case:
1,914 bytes loaded in 2.80 ms
leaves if you walk it: 3.82e+29A 1,914-byte file loaded in a few milliseconds, and the structure it describes has about 3.82e+29 leaves if anything walks it. The loader is not what breaks; the first code that touches the result is.
Aliases can also point backwards. The YAML specification says that “a sequence may contain the same node more than once. It could even contain itself.” PyYAML allows it:
loop[0] is loop: True
json.dumps -> Circular reference detectedThe list &a [*a] loads into a list whose first item is the list itself, and json.dumps refuses it with Circular reference detected. Recursive code that does not remember what it has already visited will recurse until Python raises a RecursionError. Finally, nesting without any aliases:
depth 100: loaded
depth 5000: RecursionErrorOne hundred nested lists load, and five thousand raise RecursionError. That is an ordinary Python exception rather than a crash, but a service that does not catch it returns an error page for one bad request. Remember all three behaviours for Step 8.
Step 6: Catch the key that appears twice
The YAML specification says that mapping keys must be unique: “The content of a mapping node is an unordered set of key/value node pairs, with the restriction that each of the keys is unique.” That is a rule for the file’s author, and PyYAML does not enforce it when loading. Here is what that means for a settings file:
# step06_duplicates.py - when a key appears twice, the last one wins, silently
import yaml
doc = """\
sandbox:
network: false
allowed_hosts: [pypi.org]
sandbox:
network: true
"""
print(doc, end="")
print("-> safe_load gives:", yaml.safe_load(doc))Run python step06_duplicates.py:
sandbox:
network: false
allowed_hosts: [pypi.org]
sandbox:
network: true
-> safe_load gives: {'sandbox': {'network': True}}The first sandbox block says network: false and lists an allowed host. The second sandbox block silently replaces the whole first one, so the loaded settings say network: true and the host list is gone. A reviewer who reads the top of the file sees a locked-down sandbox, while the program uses the bottom. Anyone who can add one more block to the file can change a setting without touching the lines a reviewer is looking at. Step 8 makes the loader refuse duplicates.
Step 7: See what safe_load guesses when you do not use quotes
When a value has no tag, the resolver decides its type by matching the text against patterns. PyYAML follows the older YAML 1.1 rules, and one of them is this list of boolean words from the YAML 1.1 boolean type: y|Y|yes|Yes|YES|n|N|no|No|NO|true|True|TRUE|false|False|FALSE|on|On|ON|off|Off|OFF. YAML 1.2 narrowed this. Its core schema recognises only true, True, TRUE, false, False and FALSE, and the specification calls that schema “the recommended default schema that YAML processor should use unless instructed otherwise.” PyYAML’s output for 14 plain values shows how far it still is from that:
# step07_surprises.py - what yaml.safe_load guesses from unquoted text
import yaml
SCALARS = ["NO", "yes", "on", "y", "3.10", "0755", "010", "1:30", "1_000",
"2026-11-02", "~", "1e3", "0x1F", ".inf"]
print(f"{'YAML text':<12}{'safe_load gives':<30}type")
for text in SCALARS:
value = yaml.safe_load(text)
print(f"{text:<12}{value!r:<30}{type(value).__name__}")
print()
print(yaml.safe_load("countries: [GB, NO, SE]"))
print(yaml.safe_load("on: push"))Run python step07_surprises.py:
YAML text safe_load gives type
NO False bool
yes True bool
on True bool
y 'y' str
3.10 3.1 float
0755 493 int
010 8 int
1:30 90 int
1_000 1000 int
2026-11-02 datetime.date(2026, 11, 2) date
~ None NoneType
1e3 '1e3' str
0x1F 31 int
.inf inf float
{'countries': ['GB', False, 'SE']}
{True: 'push'}Go down the first column. NO is False, the famous Norway problem: the list [GB, NO, SE] of country codes arrives as ['GB', False, 'SE'], as the second-to-last line shows. yes and on are True, so in on: push the key is the boolean True, not the text on. (y stays a string: PyYAML implements the long words but not the one-letter forms that the 1.1 page lists.) 0755 is 493 because a leading zero means octal, and 010 is 8 for the same reason, so an identifier such as a zip code written without quotes changes value. 1:30 is 90 because the colon form is base 60. 1_000 is 1000 and 0x1F is 31. 2026-11-02 becomes a date object, so a plain-text field that happens to look like a date stops being text. And 3.10 is the float 3.1, so a Python version or a release number written without quotes loses its trailing zero. That last one happens in YAML 1.1 and in 1.2 alike, so no loader can fix it; only quotes or a schema can.
Step 8: Build a loader that only returns plain data
You now have a list of problems and you know which stage each one lives in, so you can fix each in its own stage. The loader goes in one file, safeyaml.py. Put the five code blocks below into it in order; the first line of each block after the first, which says (continued), only marks where one block ends and the next begins and is not part of the file.
8a. Limits and one error type
# safeyaml.py - a YAML loader for files that someone else wrote
import re
from collections.abc import Hashable
import yaml
from yaml.constructor import ConstructorError
from yaml.nodes import ScalarNode, SequenceNode
MAX_BYTES = 64 * 1024 # refuse bigger files before parsing anything
MAX_NODES = 50_000 # budget for the document after aliases are expanded
TAG = "tag:yaml.org,2002:"
ALLOWED_TAGS = {TAG + name
for name in ("null", "bool", "int", "float", "str", "seq", "map")}
class YamlRejected(ValueError):
"""The file is not acceptable. The message says why, in plain words."""There are two budgets: a file may not exceed 64 KiB, and once aliases are expanded the document may not exceed 50,000 nodes. A node is one scalar, list or dictionary in the parsed tree, and the composer produces them. ALLOWED_TAGS lists the only seven types we accept: null, bool, int, float, str, seq (a list) and map (a dictionary). YamlRejected is the one exception your program has to catch, and it is a ValueError.
8b. Check the whole document before building any of it
# safeyaml.py (continued)
def check_nodes(root, max_nodes):
"""Walk the node graph once. Reject forbidden tags, loops and huge expansions."""
sizes = {} # id(node) -> how many nodes it has once aliases are expanded
inside = set() # ids of the nodes we are currently inside of
def walk(node):
key = id(node)
if key in sizes:
return sizes[key]
if key in inside:
raise YamlRejected("recursive alias: a node contains itself")
if node.tag not in ALLOWED_TAGS:
raise YamlRejected(f"tag {node.tag!r} is not allowed")
inside.add(key)
if isinstance(node, ScalarNode):
size = 1
elif isinstance(node, SequenceNode):
size = 1 + sum(walk(child) for child in node.value)
else:
size = 1 + sum(walk(k) + walk(v) for k, v in node.value)
inside.discard(key)
if size > max_nodes:
raise YamlRejected(f"document expands to more than {max_nodes:,} nodes")
sizes[key] = size
return size
return walk(root)This function is the heart of the defence. It walks the node graph once and does three jobs. It refuses any node whose tag is not on the list, so a Python tag is rejected here, before the constructor ever sees it. It refuses a node that contains itself, detected with the set inside, which holds the nodes we are currently in. And it works out how large each node would be if every alias were expanded, storing the answer by id(node) in sizes. That memo is what keeps it fast. The 30-level bomb has only 72 distinct nodes, so the function visits each node at most once and stops at the first level whose expanded size is over the budget. It never builds the huge structure; it only does the arithmetic.
8c. The loader class
# safeyaml.py (continued)
class StrictLoader(yaml.SafeLoader):
"""SafeLoader that also checks the whole document before building anything."""
max_nodes = MAX_NODES
def get_single_node(self):
node = super().get_single_node()
if node is not None:
check_nodes(node, self.max_nodes)
return node
def construct_mapping(self, node, deep=False):
seen = set()
for key_node, _ in node.value:
key = self.construct_object(key_node, deep=True)
if not isinstance(key, Hashable):
continue # SafeLoader reports unhashable keys itself
if key in seen:
raise ConstructorError("while constructing a mapping", node.start_mark,
f"found duplicate key {key!r}", key_node.start_mark)
seen.add(key)
return super().construct_mapping(node, deep=deep)StrictLoader starts from SafeLoader, so it begins with no Python constructors at all. get_single_node is the method PyYAML calls to get the document’s root node, so overriding it lets us run check_nodes after the document is parsed and before anything is constructed. construct_mapping runs every time a dictionary is built. We construct each key, remember the keys we have seen, and raise PyYAML’s own ConstructorError, pointing at the second key, when one repeats. Because it compares the constructed Python values, 1 and 1.0 count as the same key, which suits a settings file.
8d. Replace the type guesses
# safeyaml.py (continued)
# YAML 1.1 lets plain text turn into booleans, dates, octal and base-60 numbers.
# Replace those guesses with JSON-like ones: only true/false, null and plain numbers.
StrictLoader.yaml_implicit_resolvers = {}
for _tag, _pattern, _first in [
(TAG + "null", r"^(?:~|null|Null|NULL|)$", ["~", "n", "N", ""]),
(TAG + "bool", r"^(?:true|True|TRUE|false|False|FALSE)$", list("tTfF")),
(TAG + "int", r"^-?(?:0|[1-9][0-9]*)$", list("-0123456789")),
(TAG + "float", r"^-?(?:0|[1-9][0-9]*)"
r"(?:\.[0-9]+(?:[eE][-+]?[0-9]+)?|[eE][-+]?[0-9]+)$", list("-0123456789")),
]:
StrictLoader.add_implicit_resolver(_tag, re.compile(_pattern), _first)This is the resolver fix. Assigning an empty dictionary to yaml_implicit_resolvers throws away all of YAML 1.1’s guesses for this class, and the loop adds four back. Each rule is a tag, a regular expression and the list of first characters worth trying; PyYAML only runs a rule when the text starts with one of them. The result is deliberately close to JSON. null, ~ and an empty value are null; only true and false in three spellings are booleans; integers have no sign except a minus, no leading zeros, no underscores and no colons; floats need a dot or an exponent. Everything else is a string, including NO, 0755, 1:30 and 2026-11-02. This is narrower than the YAML 1.2 core schema, which also allows a plus sign, octal and hexadecimal integers and infinity, because a settings file rarely needs them. And because the merge-key rule is gone, << is no longer special either: it is just a string key that your schema will reject.
One detail worth knowing: the dictionary is assigned on StrictLoader itself, so plain yaml.safe_load elsewhere in your program keeps its old behaviour. A test in Step 10 checks that.
8e. The entry points
# safeyaml.py (continued)
def load_text(data):
"""Parse YAML from an untrusted source into plain Python data."""
if isinstance(data, str):
data = data.encode("utf-8")
if len(data) > MAX_BYTES:
raise YamlRejected(f"{len(data):,} bytes is over the {MAX_BYTES:,} byte limit")
try:
text = data.decode("utf-8-sig")
except UnicodeDecodeError as exc:
raise YamlRejected(f"not valid UTF-8 ({exc.reason})") from None
try:
# Ruff cannot see that StrictLoader is a SafeLoader subclass, so it objects.
return yaml.load(text, Loader=StrictLoader) # noqa: S506
except yaml.YAMLError as exc:
words = [getattr(exc, "context", None), getattr(exc, "problem", None)]
message = ", ".join(w for w in words if w) or str(exc).splitlines()[0]
mark = getattr(exc, "problem_mark", None)
where = f" (line {mark.line + 1})" if mark else ""
raise YamlRejected(message + where) from None
except RecursionError:
raise YamlRejected("nesting is too deep") from None
def load_file(path):
"""Read at most one byte past the limit, so a huge file is never read whole."""
with open(path, "rb") as handle:
return load_text(handle.read(MAX_BYTES + 1))load_text accepts text or bytes and refuses a file over the byte limit before parsing it. It then decodes the bytes as strict UTF-8; PyYAML’s own documentation says it also accepts UTF-16, which is one more way for a file to look different to a reviewer than it does to a program. Anything PyYAML complains about becomes YamlRejected with its line number when PyYAML gives one, and RecursionError becomes “nesting is too deep”. load_file reads at most one byte beyond the limit, so a multi-gigabyte file is never read into memory. The # noqa: S506 comment silences one Ruff rule on one line, and the comment above it says why; Step 11 explains the rule.
8f. Try it on every nasty document
This script sends the seven documents from Step 3, the bomb, the loop, a duplicate key, two documents in one file and deep nesting through load_text:
# step08_guard.py - the same documents through the strict loader
import textwrap
import time
import yaml
from labkit import PAYLOADS, bomb, scratch
from safeyaml import StrictLoader, YamlRejected, load_text
DOCS = {
**PAYLOADS,
"alias bomb, 30 levels": bomb(30),
"alias to itself": "&a [*a]",
"duplicate key": "a: 1\na: 2\n",
"two documents": "a: 1\n---\nb: 2\n",
"deep nesting": "[" * 5000 + "]" * 5000,
}
with scratch():
for name, text in DOCS.items():
try:
line = f"{name:<23}accepted {load_text(text)!r}"
except YamlRejected as exc:
line = f"{name:<23}REJECTED {exc}"
print(textwrap.fill(line, width=76, subsequent_indent=" " * 4))
start = time.perf_counter()
try:
load_text(bomb(30))
except YamlRejected:
pass
print(f"the 30-level bomb was refused in {(time.perf_counter() - start) * 1000:.1f} ms")
print("=== what the checks cost on an honest file")
honest = "".join(f"- {{id: {n}, name: item-{n}, tags: [a, b], ok: true}}\n" for n in range(1000))
print(f"{len(honest.encode()):,} bytes, 1000 mappings")
for label, loader in (("CSafeLoader (libyaml)", yaml.CSafeLoader),
("SafeLoader", yaml.SafeLoader), ("StrictLoader", StrictLoader)):
runs = []
for _ in range(5):
start = time.perf_counter()
yaml.load(honest, Loader=loader)
runs.append(time.perf_counter() - start)
print(f"{label:<22}{min(runs) * 1000:>7.1f} ms")Run python step08_guard.py:
plain mapping accepted {'name': 'demo', 'size': 3}
!!set REJECTED tag 'tag:yaml.org,2002:set' is not allowed
!!python/tuple REJECTED tag 'tag:yaml.org,2002:python/tuple' is not
allowed
!!python/module REJECTED tag 'tag:yaml.org,2002:python/module:os' is
not allowed
!!python/name REJECTED tag 'tag:yaml.org,2002:python/name:os.mkdir'
is not allowed
!!python/object/new REJECTED tag
'tag:yaml.org,2002:python/object/new:os.mkdir' is not allowed
!!python/object/apply REJECTED tag
'tag:yaml.org,2002:python/object/apply:os.mkdir' is not allowed
alias bomb, 30 levels REJECTED document expands to more than 50,000 nodes
alias to itself REJECTED recursive alias: a node contains itself
duplicate key REJECTED while constructing a mapping, found
duplicate key 'a' (line 2)
two documents REJECTED expected a single document in the stream,
but found another document (line 2)
deep nesting REJECTED nesting is too deep
the 30-level bomb was refused in 2.7 msEvery dangerous document is rejected with a sentence that a person can act on, and the bomb is refused in a few milliseconds without ever building its leaves. The plain mapping is still accepted. Note that !!set, which safe_load allowed, is now refused: the policy is plain data only, and a settings file has no use for sets. If you need more, add a tag to ALLOWED_TAGS on purpose.
The same script ends by timing the loaders on an honest 52 KB file of 1,000 small mappings:
51,780 bytes, 1000 mappings
CSafeLoader (libyaml) 16.2 ms
SafeLoader 114.8 ms
StrictLoader 117.2 msThe strict checks added only a few milliseconds to SafeLoader, but both pure-Python loaders took about seven times as long as libyaml’s CSafeLoader on the same file. For a settings file read once at start-up, about a tenth of a second is fine. If you parse large files in a hot path, use CSafeLoader only for files you trust.
8g. Compare three loaders side by side
One more script shows the effect of the resolver change, with ruamel.yaml as a third column. Its PyPI page describes it as “a YAML 1.2 loader/dumper package for Python”, and I used it only to see what a 1.2 loader does with the same text.
# step08b_compare.py - the same plain text through three loaders
import datetime
import yaml
from safeyaml import StrictLoader
try:
from ruamel.yaml import YAML
ruamel = YAML(typ="safe", pure=True)
except ImportError:
ruamel = None
SCALARS = ["NO", "yes", "on", "y", "3.10", "0755", "1:30", "1_000",
"2026-11-02", "~", "1e3", "0x1F", ".inf", "-0.5"]
def short(value):
if isinstance(value, datetime.date):
return f"date {value.isoformat()}"
return repr(value)
print(f"{'YAML text':<12}{'safe_load':<20}{'StrictLoader':<20}ruamel.yaml")
for text in SCALARS:
old = yaml.safe_load(text)
new = yaml.load(text, Loader=StrictLoader)
other = ruamel.load(text) if ruamel else "n/a"
print(f"{text:<12}{short(old):<20}{short(new):<20}{short(other)}")Run python step08b_compare.py:
YAML text safe_load StrictLoader ruamel.yaml
NO False 'NO' 'NO'
yes True 'yes' 'yes'
on True 'on' 'on'
y 'y' 'y' 'y'
3.10 3.1 3.1 3.1
0755 493 '0755' 755
1:30 90 '1:30' '1:30'
1_000 1000 '1_000' 1000
2026-11-02 date 2026-11-02 '2026-11-02' date 2026-11-02
~ None None None
1e3 '1e3' 1000.0 1000.0
0x1F 31 '0x1F' 31
.inf inf '.inf' inf
-0.5 -0.5 -0.5 -0.5The strict loader keeps NO, yes, on, 0755, 1:30, 1_000, 0x1F, .inf and the date as text, and keeps the JSON-style numbers. The YAML 1.2 loader fixes the boolean words and the base-60 number, but it still reads 0755 as the integer 755, 1_000 as 1000 and the date as a date. And all three turn 3.10 into 3.1, which is why the next step adds a schema.
Step 9: Check the shape of the data with a schema
A loader can only decide what a piece of text is; it cannot decide what your program expects. For that you need a schema, a description of the allowed keys and their types. Pydantic is a library that builds one from a Python class. The class below describes a plugin manifest. It rejects keys it does not know (extra="forbid") and accepts only a real boolean for enabled (StrictBool).
# manifest.py - the shape of a plugin manifest, checked after the YAML is loaded
from datetime import date
from pydantic import BaseModel, ConfigDict, Field, StrictBool, ValidationError
from safeyaml import YamlRejected, load_file
class Plugin(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str = Field(pattern=r"^[a-z][a-z0-9-]{1,40}$")
version: str
python_version: str
enabled: StrictBool = False
countries: list[str] = []
released: date | None = None
def describe(exc):
"""One short sentence per problem, with the value that was found."""
parts = []
for error in exc.errors():
where = ".".join(str(p) for p in error["loc"]) or "document"
parts.append(f"{where}: {error['msg']} (got {error['input']!r:.30})")
return "; ".join(parts)
def load_plugin(path):
data = load_file(path)
try:
return Plugin.model_validate(data)
except ValidationError as exc:
raise YamlRejected(describe(exc)) from NoneThe important design choice is that version and python_version are declared as strings. Pydantic’s conversion table says that a number is accepted as a string “Only if coerce_numbers_to_str is enabled.” We do not enable it, so an unquoted 3.10, which the loader reads as the float 3.1, is refused with a message that shows the author what was found, instead of being silently stored as 3.1. The date is the opposite case: released is declared as a date, and Pydantic parses the text 2026-11-02 into one, so the resolver can leave dates as text while the schema turns them into real dates where you want them.
Now six small files, one good and five bad. The script writes them into a scratch folder and loads each with load_plugin:
# step09_files.py - six files, one loader
import os
import textwrap
from pathlib import Path
from labkit import bomb, scratch
from manifest import load_plugin
from safeyaml import YamlRejected
FILES = {
"good.yaml": ('name: geo-fence\nversion: "1.10"\npython_version: "3.10"\n'
'enabled: true\ncountries: [GB, NO, SE]\nreleased: 2026-11-02\n'),
"sloppy.yaml": ("name: geo-fence\nversion: 1.10\npython_version: 3.10\n"
"enabled: yes\ncountries: [GB, NO, SE]\n"),
"tagged.yaml": ('name: !!python/object/apply:os.mkdir ["CANARY_RAN"]\n'
'version: "1"\npython_version: "3.13"\n'),
"bomb.yaml": bomb(30),
"dupes.yaml": 'name: geo-fence\nversion: "1"\nversion: "2"\npython_version: "3.13"\n',
"extra.yaml": 'name: geo-fence\nversion: "1"\npython_version: "3.13"\nrun_on_install: true\n',
}
with scratch():
for name, text in FILES.items():
Path(name).write_text(text, encoding="utf-8", newline="\n")
try:
plugin = load_plugin(name)
verdict = (f"accepted {plugin.name} {plugin.version} "
f"py{plugin.python_version} {plugin.countries}")
except YamlRejected as exc:
verdict = f"REJECTED {exc}"
print(textwrap.fill(f"{name:<12}{verdict}", width=76, subsequent_indent=" " * 12))
print("CANARY_RAN created:", os.path.exists("CANARY_RAN"))Run python step09_files.py:
good.yaml accepted geo-fence 1.10 py3.10 ['GB', 'NO', 'SE']
sloppy.yaml REJECTED version: Input should be a valid string (got 1.1);
python_version: Input should be a valid string (got 3.1);
enabled: Input should be a valid boolean (got 'yes')
tagged.yaml REJECTED tag 'tag:yaml.org,2002:python/object/apply:os.mkdir'
is not allowed
bomb.yaml REJECTED document expands to more than 50,000 nodes
dupes.yaml REJECTED while constructing a mapping, found duplicate key
'version' (line 3)
extra.yaml REJECTED run_on_install: Extra inputs are not permitted (got
True)
CANARY_RAN created: FalseThe good file loads with its versions still exact strings and with NO intact in the country list. The sloppy file, which is what a hurried author would write, produces three precise messages: the author sees got 1.1 and got 3.1 and learns to quote the values. The file with a Python tag, the bomb and the duplicate key are refused by the loader, and the file with an unknown key is refused by the schema. The last line shows that the canary folder never appeared.
Step 10: Test every promise, then break the loader on purpose
A security control that is not tested tends to stop working without anyone noticing. This file has one test per promise made above. A few details are worth knowing. monkeypatch.chdir moves each Python-tag test into pytest’s own temporary folder, so a failing test can never create the canary folder in your project. The bomb test also asserts that the refusal takes under a second. And test_global_safe_load_is_untouched proves that building StrictLoader did not change yaml.safe_load for the rest of your program.
# test_safeyaml.py - the promises of safeyaml.py and manifest.py, one test each
import datetime
import time
import pytest
import yaml
from labkit import bomb
from manifest import load_plugin
from safeyaml import MAX_BYTES, StrictLoader, YamlRejected, load_file, load_text
PYTHON_TAGS = [
'!!python/object/apply:os.mkdir ["CANARY_RAN"]',
'!!python/object/new:os.mkdir ["CANARY_RAN"]',
"!!python/name:os.mkdir",
"!!python/module:os",
"!!python/tuple [1, 2]",
]
GOOD = ('name: geo-fence\nversion: "1.10"\npython_version: "3.10"\n'
'enabled: true\ncountries: [GB, NO, SE]\nreleased: 2026-11-02\n')
SLOPPY = ("name: geo-fence\nversion: 1.10\npython_version: 3.10\n"
"enabled: yes\ncountries: [GB, NO, SE]\n")
def test_plain_data_loads():
text = "name: demo\nsizes: [1, 2.5, true, null]\n"
assert load_text(text) == {"name": "demo", "sizes": [1, 2.5, True, None]}
@pytest.mark.parametrize("text", PYTHON_TAGS)
def test_python_tags_are_refused(text, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with pytest.raises(YamlRejected, match="not allowed"):
load_text(text)
assert not (tmp_path / "CANARY_RAN").exists()
@pytest.mark.parametrize("text", ["!!binary aGk=", "!!set {a, b}",
"!!timestamp 2026-11-02", "!!omap [a: 1]"])
def test_other_explicit_tags_are_refused(text):
with pytest.raises(YamlRejected, match="not allowed"):
load_text(text)
def test_no_python_constructors_are_registered():
assert not [tag for tag in StrictLoader.yaml_constructors if tag and "python" in tag]
assert not StrictLoader.yaml_multi_constructors
def test_alias_bomb_is_refused_quickly():
start = time.perf_counter()
with pytest.raises(YamlRejected, match="expands to more than"):
load_text(bomb(30))
assert time.perf_counter() - start < 1.0
def test_modest_alias_use_is_allowed():
data = load_text("defaults: &d {retries: 3}\nprod: *d\nstaging: *d\n")
assert data["prod"] == data["staging"] == {"retries": 3}
def test_recursive_alias_is_refused():
with pytest.raises(YamlRejected, match="recursive"):
load_text("&a [*a]\n")
def test_duplicate_key_is_refused_with_its_line_number():
text = "sandbox:\n network: false\n hosts: [a]\nsandbox:\n network: true\n"
with pytest.raises(YamlRejected, match=r"duplicate key 'sandbox' \(line 4\)"):
load_text(text)
def test_duplicate_key_in_a_nested_mapping_is_refused():
with pytest.raises(YamlRejected, match="duplicate key 'a'"):
load_text("outer:\n a: 1\n a: 2\n")
def test_same_key_in_different_mappings_is_fine():
assert load_text("x: {a: 1}\ny: {a: 2}\n") == {"x": {"a": 1}, "y": {"a": 2}}
def test_merge_key_is_an_ordinary_string():
data = load_text("base: &b {a: 1}\nchild:\n <<: *b\n c: 2\n")
assert data["child"] == {"<<": {"a": 1}, "c": 2}
@pytest.mark.parametrize("text", ["NO", "no", "yes", "Yes", "on", "Off", "y", "n"])
def test_yaml11_words_stay_strings(text):
assert load_text(text) == text
def test_the_norway_list():
assert load_text("countries: [GB, NO, SE]") == {"countries": ["GB", "NO", "SE"]}
def test_booleans_and_null_use_the_json_words():
assert load_text("[true, True, FALSE, false, null, ~]") == [True, True, False, False, None, None]
assert load_text("a:\n") == {"a": None}
@pytest.mark.parametrize("text", ["0755", "010", "1:30", "1_000", "0x1F", "+1",
".inf", "2026-11-02", "0o17"])
def test_odd_numbers_and_dates_stay_strings(text):
assert load_text(text) == text
def test_json_style_numbers_still_work():
assert load_text("[0, -3, 42, 2.5, 1e3, -0.5]") == [0, -3, 42, 2.5, 1000.0, -0.5]
def test_version_like_numbers_are_still_numbers():
assert load_text("3.10") == 3.1 # no loader can know you meant a string
def test_size_limit_boundary():
ok = "a: " + "x" * (MAX_BYTES - 4) + "\n"
assert len(ok) == MAX_BYTES
assert load_text(ok)["a"] == "x" * (MAX_BYTES - 4)
with pytest.raises(YamlRejected, match="byte limit"):
load_text(ok + "#")
def test_load_file_refuses_a_huge_file(tmp_path):
path = tmp_path / "big.yaml"
path.write_bytes(b"#" * (MAX_BYTES * 4))
with pytest.raises(YamlRejected, match="byte limit"):
load_file(path)
def test_utf16_and_bad_bytes_are_refused():
with pytest.raises(YamlRejected, match="UTF-8"):
load_text("a: 1".encode("utf-16"))
with pytest.raises(YamlRejected, match="UTF-8"):
load_text(b"a: \xff\xfe\n")
def test_two_documents_are_refused():
with pytest.raises(YamlRejected, match="single document"):
load_text("a: 1\n---\nb: 2\n")
def test_deep_nesting_is_refused_not_crashed():
with pytest.raises(YamlRejected, match="too deep"):
load_text("[" * 5000 + "]" * 5000)
def test_syntax_error_names_a_line():
with pytest.raises(YamlRejected, match=r"\(line \d+\)"):
load_text("a: 1\nb: [1, 2\n")
def test_global_safe_load_is_untouched():
assert yaml.safe_load("NO") is False
assert yaml.safe_load("2026-11-02") == datetime.date(2026, 11, 2)
def test_good_manifest_loads(tmp_path):
path = tmp_path / "plugin.yaml"
path.write_text(GOOD, encoding="utf-8")
plugin = load_plugin(path)
assert (plugin.version, plugin.python_version) == ("1.10", "3.10")
assert plugin.countries == ["GB", "NO", "SE"]
assert plugin.released.isoformat() == "2026-11-02"
def test_sloppy_manifest_names_every_problem(tmp_path):
path = tmp_path / "plugin.yaml"
path.write_text(SLOPPY, encoding="utf-8")
with pytest.raises(YamlRejected) as info:
load_plugin(path)
message = str(info.value)
assert "version: Input should be a valid string (got 1.1)" in message
assert "python_version: Input should be a valid string (got 3.1)" in message
assert "enabled: Input should be a valid boolean (got 'yes')" in message
def test_unknown_manifest_keys_are_refused(tmp_path):
path = tmp_path / "plugin.yaml"
path.write_text(GOOD + "run_on_install: true\n", encoding="utf-8")
with pytest.raises(YamlRejected, match="run_on_install: Extra inputs are not permitted"):
load_plugin(path)Run python -m pytest -q:
................................................. [100%]
49 passed in 0.71sPassing tests only show that the code does what the tests say. To find out whether the tests would notice a mistake, mutation testing changes the code on purpose and checks that some test fails. This script copies the files to a temporary folder, applies one of ten small changes (accept any tag, drop the expansion budget, allow duplicate keys, keep the YAML 1.1 guesses, build on UnsafeLoader and so on) and runs pytest, expecting a failure each time. It also asserts that each pattern occurs exactly once, so a typo cannot turn a mutant into a no-op.
# mutations.py - break the loader on purpose and make sure the tests notice
import pathlib
import re
import shutil
import subprocess
import sys
import tempfile
HERE = pathlib.Path(__file__).parent
FILES = ["safeyaml.py", "manifest.py", "labkit.py", "test_safeyaml.py"]
MUTANTS = [
("accept any tag", "safeyaml.py", "if node.tag not in ALLOWED_TAGS:", "if False:"),
("no expansion budget", "safeyaml.py", "if size > max_nodes:", "if False:"),
("allow recursive alias", "safeyaml.py", "if key in inside:", "if False:"),
("allow duplicate keys", "safeyaml.py", "if key in seen:", "if False:"),
("keep YAML 1.1 guesses", "safeyaml.py",
"StrictLoader.yaml_implicit_resolvers = {}", "pass"),
("size limit off by one", "safeyaml.py",
"if len(data) > MAX_BYTES:", "if len(data) >= MAX_BYTES:"),
("build on UnsafeLoader", "safeyaml.py",
"class StrictLoader(yaml.SafeLoader):", "class StrictLoader(yaml.UnsafeLoader):"),
("lenient UTF-8", "safeyaml.py",
'data.decode("utf-8-sig")', 'data.decode("utf-8", errors="replace")'),
("ignore unknown keys", "manifest.py", 'extra="forbid"', 'extra="ignore"'),
("lax booleans", "manifest.py", "enabled: StrictBool", "enabled: bool"),
]
def run(mutant):
with tempfile.TemporaryDirectory() as tmp:
for name in FILES:
shutil.copy(HERE / name, tmp)
if mutant:
_, name, old, new = mutant
path = pathlib.Path(tmp) / name
text = path.read_text(encoding="utf-8")
assert text.count(old) == 1, (name, old, text.count(old))
path.write_text(text.replace(old, new), encoding="utf-8", newline="\n")
result = subprocess.run(
[sys.executable, "-m", "pytest", "-q", "-x", "-p", "no:cacheprovider", tmp],
capture_output=True, text=True)
failed = re.search(r"^FAILED \S+?::(\w+)", result.stdout, re.M)
return result.returncode, failed.group(1) if failed else ""
code, _ = run(None)
print(f"{'unchanged code':<24}{'tests pass' if code == 0 else 'TESTS FAIL'}")
caught = 0
for mutant in MUTANTS:
code, test = run(mutant)
caught += code != 0
print(f"{mutant[0]:<24}{'caught by ' + test if code else 'NOT CAUGHT'}")
print(f"{caught} of {len(MUTANTS)} planted bugs were caught")Run python mutations.py; it takes a few seconds per mutant:
unchanged code tests pass
accept any tag caught by test_python_tags_are_refused
no expansion budget caught by test_alias_bomb_is_refused_quickly
allow recursive alias caught by test_recursive_alias_is_refused
allow duplicate keys caught by test_duplicate_key_is_refused_with_its_line_number
keep YAML 1.1 guesses caught by test_merge_key_is_an_ordinary_string
size limit off by one caught by test_size_limit_boundary
build on UnsafeLoader caught by test_no_python_constructors_are_registered
lenient UTF-8 caught by test_utf16_and_bad_bytes_are_refused
ignore unknown keys caught by test_unknown_manifest_keys_are_refused
lax booleans caught by test_sloppy_manifest_names_every_problem
10 of 10 planted bugs were caughtAll ten were caught. If one had survived, it would point to a promise with no test, and you would write that test before trusting the loader. Notice that the mutant that builds on UnsafeLoader is caught by the test that inspects the registered constructors, not by the tag-refusal tests, because the tag allow-list would still have refused the Python tags. That test is what proves the base class itself is safe: two layers, each checked by its own test.
Step 11: Stop the next unsafe call before it merges
Someone will eventually write yaml.load(text, Loader=yaml.Loader) again, perhaps pasted from an old answer, perhaps written by a coding assistant. A linter is a tool that reads source code and reports suspicious patterns. Ruff has a rule for this, S506, and its documentation explains the risk: “Running the yaml.load function over untrusted YAML files is insecure, as yaml.load allows for the creation of arbitrary Python objects, which can then be used to execute arbitrary code.” A rule is only as good as its coverage, so measure it. This file lists 14 ways to call the loader, with the verdict I expect in a comment on each line:
# gate_samples.py - fourteen ways to call the YAML loader (read by step11_gate.py, never run)
import yaml
import yaml as y
from safeyaml import StrictLoader
from yaml import FullLoader as FL
from yaml import Loader, SafeLoader, load
from yaml import load as yload
chosen_loader = yaml.SafeLoader
text = "a: 1"
yaml.safe_load(text) # expect: ok
yaml.load(text, Loader=yaml.SafeLoader) # expect: ok
load(text, Loader=SafeLoader) # expect: ok
yaml.load(text, Loader=StrictLoader) # expect: ok
yaml.load(text) # expect: bad
yaml.load(text, Loader=yaml.Loader) # expect: bad
yaml.load(text, yaml.FullLoader) # expect: bad
yaml.full_load(text) # expect: bad
yaml.unsafe_load(text) # expect: bad
y.load(text, Loader=y.UnsafeLoader) # expect: bad
load(text, Loader=Loader) # expect: bad
yload(text, Loader=FL) # expect: bad
yaml.load_all(text, Loader=yaml.FullLoader) # expect: bad
yaml.load(text, Loader=chosen_loader) # expect: badNext comes a gate written with Python’s standard ast module. An AST (abstract syntax tree) is the structure Python builds from source code before running it, and walking it lets you look at calls instead of searching for text. The gate treats the safe loaders as an allow-list: any yaml.load call whose Loader is not on the list is a finding, and so is any call to full_load or unsafe_load. It follows aliases such as import yaml as y and from yaml import load as yload.
# yamlgate.py - fail the build when YAML is loaded with a loader that can build objects
import ast
import sys
from pathlib import Path
SAFE_LOADERS = {"SafeLoader", "CSafeLoader", "BaseLoader", "CBaseLoader", "StrictLoader"}
OBJECT_BUILDERS = {"unsafe_load", "unsafe_load_all", "full_load", "full_load_all"}
GENERIC = {"load", "load_all"}
def loader_name(expr, modules, imported):
"""Name of the Loader expression, or None when we cannot tell."""
if (isinstance(expr, ast.Attribute) and isinstance(expr.value, ast.Name)
and expr.value.id in modules):
return expr.attr
if isinstance(expr, ast.Name):
return imported.get(expr.id, expr.id)
return None
def check_source(source):
"""Return sorted (line, reason) pairs for every risky YAML call in `source`."""
tree = ast.parse(source)
modules, functions, imported = set(), {}, {}
for node in ast.walk(tree):
if isinstance(node, ast.Import):
modules.update(a.asname or a.name for a in node.names if a.name == "yaml")
elif isinstance(node, ast.ImportFrom) and node.module == "yaml":
for alias in node.names:
local = alias.asname or alias.name
if alias.name in OBJECT_BUILDERS | GENERIC:
functions[local] = alias.name
else:
imported[local] = alias.name
found = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
func = node.func
if (isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name)
and func.value.id in modules):
name = func.attr
elif isinstance(func, ast.Name) and func.id in functions:
name = functions[func.id]
else:
continue
if name in OBJECT_BUILDERS:
found.append((node.lineno, f"yaml.{name} can build Python objects"))
elif name in GENERIC:
arg = next((kw.value for kw in node.keywords if kw.arg == "Loader"), None)
if arg is None and len(node.args) > 1:
arg = node.args[1]
used = loader_name(arg, modules, imported) if arg is not None else None
if used not in SAFE_LOADERS:
found.append((node.lineno, f"yaml.{name} with Loader={used}"))
return sorted(found)
def main(paths):
bad = 0
for root in paths:
root = Path(root)
for path in sorted(root.rglob("*.py")) if root.is_dir() else [root]:
for line, why in check_source(path.read_text(encoding="utf-8")):
print(f"{path}:{line}: {why}")
bad += 1
return 1 if bad else 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:] or ["."]))And a script that runs Ruff’s S506 and the gate over the sample file and prints both verdicts side by side:
# step11_gate.py - what Ruff's S506 rule and our own gate each flag
import re
import subprocess
import sys
from pathlib import Path
from yamlgate import check_source
def ruff_lines(path):
result = subprocess.run(
[sys.executable, "-m", "ruff", "check", "--select", "S506", "--isolated",
"--no-cache", "--output-format", "concise", path],
capture_output=True, text=True)
return {int(m.group(1)) for m in re.finditer(r":(\d+):\d+: S506", result.stdout)}
source = Path("gate_samples.py").read_text(encoding="utf-8")
ruff = ruff_lines("gate_samples.py")
ours = {line for line, _ in check_source(source)}
print(f"{'line':<5}{'code':<45}{'expect':<8}{'Ruff':<6}gate")
bad_lines, ok_lines = [], []
for number, row in enumerate(source.splitlines(), 1):
if "# expect:" not in row:
continue
code, expect = (part.strip() for part in row.split("# expect:"))
(bad_lines if expect == "bad" else ok_lines).append(number)
print(f"{number:<5}{code:<45}{expect:<8}"
f"{'FLAG' if number in ruff else '-':<6}{'FLAG' if number in ours else '-'}")
for label, flagged in (("Ruff S506", ruff), ("our gate", ours)):
missed = [n for n in bad_lines if n not in flagged]
false_alarms = [n for n in ok_lines if n in flagged]
print(f"{label:<10}: missed {len(missed)} of {len(bad_lines)} bad lines, "
f"{len(false_alarms)} false alarms on {len(ok_lines)} ok lines")
own = ruff_lines("safeyaml.py")
print("Ruff S506 on safeyaml.py:", f"flags line(s) {sorted(own)}" if own else "clean")
print("our gate on safeyaml.py :", check_source(Path("safeyaml.py").read_text(encoding="utf-8")) or "clean")Run python step11_gate.py:
line code expect Ruff gate
12 yaml.safe_load(text) ok - -
13 yaml.load(text, Loader=yaml.SafeLoader) ok - -
14 load(text, Loader=SafeLoader) ok - -
15 yaml.load(text, Loader=StrictLoader) ok FLAG -
16 yaml.load(text) bad FLAG FLAG
17 yaml.load(text, Loader=yaml.Loader) bad FLAG FLAG
18 yaml.load(text, yaml.FullLoader) bad FLAG FLAG
19 yaml.full_load(text) bad - FLAG
20 yaml.unsafe_load(text) bad - FLAG
21 y.load(text, Loader=y.UnsafeLoader) bad FLAG FLAG
22 load(text, Loader=Loader) bad FLAG FLAG
23 yload(text, Loader=FL) bad FLAG FLAG
24 yaml.load_all(text, Loader=yaml.FullLoader) bad - FLAG
25 yaml.load(text, Loader=chosen_loader) bad FLAG FLAG
Ruff S506 : missed 3 of 10 bad lines, 1 false alarms on 4 ok lines
our gate : missed 0 of 10 bad lines, 0 false alarms on 4 ok lines
Ruff S506 on safeyaml.py: clean
our gate on safeyaml.py : cleanBoth tools flagged the dangerous yaml.load shapes, including the aliased ones. The differences are the useful part. On Ruff 0.17.3, S506 alone missed three of the ten bad lines: the calls to full_load, unsafe_load and load_all. That fits its description, which says it “Checks for uses of the yaml.load function.” It also flagged Loader=StrictLoader, a false alarm, because a linter cannot tell that a class imported from another module inherits from SafeLoader. That is why safeyaml.py carries a # noqa: S506 on its single yaml.load line, and why the last two lines show Ruff and the gate both clean on our own code. The gate flagged all ten bad lines and none of the four safe ones, because it asks a different question: is the loader on my list of known-safe names? A variable such as chosen_loader is flagged, since the gate cannot prove what it holds.
To use it in CI, run both commands against your source folder and let a non-zero exit code fail the job:
python -m ruff check --select S506 src
python yamlgate.py srcyamlgate.py exits with 1 when it prints any finding. Its allow-list matches loader names, not classes, so a class you name SafeLoader yourself would pass. Keep the list short and review every change to it.
Step 12: Confirm the whole thing works end to end
Run these four commands from your lab folder. Together they answer the questions that matter:
python -m pytest -q | tail -n 1
python mutations.py | tail -n 1
python yamlgate.py safeyaml.py manifest.py; echo "gate exit code: $?"
python -m ruff check --select S506 --isolated safeyaml.py manifest.pyYou should see all 49 tests passing, all ten planted bugs caught, no gate findings with exit code 0, and Ruff reporting that all checks passed:
49 passed in 0.72s
10 of 10 planted bugs were caught
gate exit code: 0
All checks passed!If all four lines match, your program can read a stranger’s YAML file without running anything the file names, without being flattened by a bomb, without trusting a duplicated key and without turning NO into False.
Common mistakes and gotchas
- Other loader names.
yaml.load(text, Loader=yaml.FullLoader),yaml.full_loadandyaml.unsafe_loadare not the fix. Useyaml.safe_load, orload_textfrom this tutorial for files you do not control. - Safe is not bounded.
safe_loadstops code execution, not resource use. Cap the file size before parsing, count nodes after aliases, and be careful what you do with an aliased structure afterwards. - Quote what must stay text. In your own YAML write
"NO","3.10"and"0755"with quotes whenever the value is a name, a version or an identifier. - What
StrictLoaderdoes not support. It has no merge keys, sets or timestamps, and it accepts exactly one document. For a trusted CI file that needs merge keys, usesafe_loadthere and keep the strict loader for outside input. - Speed. The strict loader is built on the pure-Python
SafeLoader. I did not test whether the same checks can hook into the libyaml-basedCSafeLoader, so do not assume it. - A schema only checks what you declare. Keep
extra="forbid", or an unknown key will pass through unnoticed. - Other languages have the same feature. The YAML specification says that processors “typically provide some form of local tags that map directly to the language’s native data structures” and gives
!ruby/object:Setas an example. Check what the parser you use does with such tags.
Where to go next
Validate every file format at the point where it enters your program: the schema step works the same way for JSON and TOML. If you need YAML 1.2 semantics, try ruamel.yaml and run the Step 8 comparison on your own sample files. Run mutations.py after every change to the loader, and add a test for any new tag you allow. For the sibling problems, read our tutorials on loading pickle and model files safely, parsing untrusted XML and stopping decompression bombs.
No Comment! Be the first one.