Path Traversal in Python (pathlib, tarfile)

Path joins and string-prefix checks do not confine file access, and resolving a path before opening it leaves a symlink race. This Python guide serves only authorized, server-issued UUIDs through a held directory descriptor, and imports regular tar files under generated names with compressed-size, entry-count, per-file, and total-byte limits. Python 3.13 and earlier used this fully trusted behavior by default; Python 3.14 defaults to the safer data filter. The filter parameter was added in Python 3.12 (backported to 3.8.17); on older releases, omit it to demonstrate the same unsafe default.^{[1], [12], [13], [8], [3], [6], [10], [11]}

💡 Plain English Explainer (ELI5)

Imagine a clerk assigned one locked cabinet. A visitor cannot choose a shelf by writing a path on a note: the service checks the visitor’s permission, gives the clerk a server-issued file token, and the clerk uses a key that stays attached to that cabinet. Archive contents go into a fresh private cabinet, receive new labels, and must fit strict size and count limits.^{[1], [8], [6]}

Core Concepts & Subsystem Terms

os.path.join() with an absolute part: If any later argument is absolute, everything before it is discarded: os.path.join("/srv/uploads", "/etc/hostname") returns /etc/hostname. .. segments are kept as written, so join never confines a path.^{[12], [1]}
String prefix checks: path.startswith("/srv/uploads") compares characters, not folders, so /srv/uploads-old/secret.txt passes. Containment has to compare whole path components.^{[13], [5]}
Directory descriptors and TOCTOU: A held directory descriptor anchors a relative os.open(..., dir_fd=...) operation to the directory opened at startup. With a server-generated single-component UUID name and O_NOFOLLOW, the request cannot replace the directory or race a later path lookup. The directory must remain service-owned.^{[7], [8]}
Bounded tar import: Python extraction filters reject unsafe paths and links, but they are not resource limits; failed extraction may also leave partial output. Stream entries, cap compressed bytes, count, per-entry bytes and total expanded bytes, and write only regular file data to generated names in private staging. Python 3.13 and earlier used this fully trusted behavior by default; Python 3.14 defaults to the safer data filter. The filter parameter was added in Python 3.12 (backported to 3.8.17); on older releases, omit it to demonstrate the same unsafe default.^{[6], [10], [11]}
Descriptor-relative file open: os.open(name, flags, dir_fd=root_fd) resolves a relative name from an already-open directory descriptor. A single server-generated component plus O_NOFOLLOW avoids request-controlled traversal and final-component symlink following; use only on supported POSIX systems.^{[7], [8]}

Framework mappings

CWE reference for the weakness described in this guide: CWE-22

Step-by-Step Attack Flow

Step 1

Request value reaches a file API

A Flask view passes request.args["file"], a form field or an uploaded archive's member names to send_file, open or extractall. Python 3.13 and earlier used this fully trusted behavior by default; Python 3.14 defaults to the safer data filter. The filter parameter was added in Python 3.12 (backported to 3.8.17); on older releases, omit it to demonstrate the same unsafe default.^{[3], [6], [10], [11]}

Step 2

The value leaves the intended folder

It starts with /, contains ../ segments, or names a sibling folder that shares the base prefix. The benign test value ../../etc/hostname illustrates the read case.^{[12], [13], [1]}

Step 3

The operating system opens the resolved path

The kernel resolves every .. and symlink and opens the file it lands on, with the web process's permissions.^{[7], [8]}

Step 4

The attacker reads or overwrites files

Reads expose .env files, keys and source code; writes from an archive can replace templates or modules that the application later runs.^{[1], [5], [4]}

Source Code: Flaw vs. Secure Implementation

UNPATCHED FLAW

# files.py: each view lets the request pick a path outside UPLOAD_DIR
import os
import tarfile

from flask import Flask, abort, request, send_file

app = Flask(__name__)
UPLOAD_DIR = "/srv/uploads"


# an absolute file name replaces UPLOAD_DIR; ../ walks out of it
@app.get("/download")
def download():
    return send_file(os.path.join(UPLOAD_DIR, request.args.get("file", "")))


# startswith() also accepts /srv/uploads-old/...
@app.get("/preview")
def preview():
    path = os.path.abspath(os.path.join(UPLOAD_DIR, request.args.get("file", "")))
    if not path.startswith(UPLOAD_DIR):
        abort(404)
    with open(path, encoding="utf-8") as f:
        return f.read(2000)


# member names such as ../../app/settings.py are written as-is
def import_archive(archive, dest):
    with tarfile.open(archive) as tar:
        # <=3.13 defaulted to fully trusted; 3.14 uses data. filter= arrived in 3.12 (3.8.17 backport).
        tar.extractall(dest, filter="fully_trusted")

HARDENED SECURE PATCH

# files.py: descriptor-relative reads and bounded, generated-name import
import os
import stat
import tarfile
import uuid
from pathlib import Path

from flask import Flask, abort, request, send_file
from flask_login import current_user, login_required  # Flask-Login integration

app = Flask(__name__)
UPLOAD_DIR = Path("/srv/uploads")  # configured, service-owned directory
if (os.open not in os.supports_dir_fd or not hasattr(os, "O_NOFOLLOW")
        or not hasattr(os, "O_DIRECTORY") or not hasattr(os, "O_CLOEXEC")):
    raise RuntimeError("this implementation requires POSIX O_NOFOLLOW/O_DIRECTORY")
UPLOAD_FD = os.open(UPLOAD_DIR, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)

MAX_ARCHIVE_BYTES = 25_000_000
MAX_ENTRY_BYTES = 10_000_000
MAX_TOTAL_BYTES = 100_000_000
MAX_ENTRIES = 1_000
CHUNK_BYTES = 64 * 1024


def open_upload(file_id: uuid.UUID) -> int:
    # One generated component, opened beneath a held directory descriptor.
    fd = os.open(file_id.hex + ".bin", os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC,
                 dir_fd=UPLOAD_FD)
    if not stat.S_ISREG(os.fstat(fd).st_mode):
        os.close(fd)
        abort(404)
    return fd


def require_file_access(user, file_id: uuid.UUID) -> None:
    # Application repository must enforce owner/tenant authorization.
    if not file_repository.user_can_read(user.id, file_id):
        abort(404)


@app.get("/download/<uuid:file_id>")
@login_required
def download(file_id: uuid.UUID):
    require_file_access(current_user, file_id)
    return send_file(os.fdopen(open_upload(file_id), "rb"),
                     download_name="download.bin", conditional=False)


@app.get("/preview/<uuid:file_id>")
@login_required
def preview(file_id: uuid.UUID):
    require_file_access(current_user, file_id)
    with os.fdopen(open_upload(file_id), "rb") as source:
        return source.read(2_001).decode("utf-8")


# `archive` is a service-owned spool file; `staging_fd` is a fresh private
# directory descriptor that untrusted users/processes cannot mutate.
def import_archive(archive: Path, staging_fd: int) -> list[str]:
    if archive.stat().st_size > MAX_ARCHIVE_BYTES:
        raise ValueError("archive too large")

    created: list[str] = []
    total = 0
    count = 0
    try:
        # Stream compressed tar input; member names are metadata, never output paths.
        with tarfile.open(archive, mode="r|*") as tar:
            for member in tar:
                count += 1
                if count > MAX_ENTRIES:
                    raise ValueError("too many archive entries")
                if not member.isfile():  # discard directories, links, and special files
                    continue
                if member.size < 0 or member.size > MAX_ENTRY_BYTES:
                    raise ValueError("archive entry too large")
                total += member.size
                if total > MAX_TOTAL_BYTES:
                    raise ValueError("archive expands beyond limit")

                source = tar.extractfile(member)
                if source is None:
                    raise tarfile.ReadError("regular entry has no data")
                name = uuid.uuid4().hex + ".bin"
                fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL |
                             os.O_NOFOLLOW | os.O_CLOEXEC, 0o600, dir_fd=staging_fd)
                created.append(name)
                with source, os.fdopen(fd, "wb") as target:
                    remaining = member.size
                    while remaining:
                        chunk = source.read(min(CHUNK_BYTES, remaining))
                        if not chunk:
                            raise tarfile.ReadError("truncated archive entry")
                        target.write(chunk)
                        remaining -= len(chunk)
        return created
    except Exception:
        # staging_fd must reference this import's private directory.
        for name in created:
            os.unlink(name, dir_fd=staging_fd)
        raise

Engineering & System Hardening Checklist

✓Use an authenticated UUID route and enforce owner/tenant authorization before opening the object; UUIDs are identifiers, not access control.^{[9], [3]}
✓Open the generated one-component filename relative to a held directory descriptor with O_NOFOLLOW; fail closed on platforms without the required descriptor-relative flags.^{[7], [8]}
✓Stream archives with a compressed-input cap, entry-count cap, per-file cap, and total expanded-byte cap. Accept only regular files, ignore archive names for output, and create generated names with O_EXCL in a fresh private staging directory.^{[6], [10], [4]}
✓Delete every staged output if extraction fails, and publish staged files only after the complete import passes validation.
✓Keep upload and staging directories service-owned and non-writable by untrusted users; run archive parsing in a worker with memory and CPU limits.^{[1], [5]}