Files, paths and safe writes

pathlib, encodings, atomic replace, file modes and untrusted paths.

Almost every automation job reads a file, writes a file, or takes a file name from someone else. This lesson covers the parts that go wrong in production: paths built from strings, logs that are not valid UTF-8, reports left half-written by a crash, files that other users can read, and a file name that walks out of the directory you meant to serve. You will use pathlib, pass encoding= and errors= on purpose, write reports with a temporary file and os.replace, check the mode a new file really gets, and contain untrusted paths with resolve() and is_relative_to().

Paths are objects

The lab works in ~/ps-files on a small tree of logs; unpack the lesson files in your home directory, or create the directory and the files as they appear. This script builds the tree, and the upload area the last section uses, so you can run it again whenever you want a clean copy:

#!/usr/bin/env bash
# Creates the practice trees for the files lesson. Safe to run again.
set -euo pipefail
rm -rf logs_demo uploads
mkdir -p logs_demo/app logs_demo/nginx uploads/2026
printf '%s\n' \
  '203.0.113.42 - - "GET /login HTTP/1.1" 200' \
  '198.51.100.7 - - "GET /admin HTTP/1.1" 403' \
  '203.0.113.42 - - "POST /login HTTP/1.1" 401' >logs_demo/nginx/access.log
printf '%s\n' 'INFO started' 'ERROR db timeout' 'ERROR db timeout' >logs_demo/app/errors.log
printf '%s\n' 'INFO started' 'INFO healthy' >logs_demo/app/service.log
# Written in Latin-1, as an old appliance would: byte 0xe9 is "é" there.
printf 'INFO user caf\xe9 logged in\nERROR disk full\n' >logs_demo/app/legacy.log
printf '%s\n' 'rotate weekly' >logs_demo/notes.txt
# The upload area for the last section.
printf '%s\n' 'quarterly access report' >uploads/report.txt
printf '%s\n' 'audit entries' >uploads/2026/audit.log

legacy.log is written the way an old appliance might write it: in Latin-1, where the byte 0xe9 means "é". The next section needs that byte. The pathlib module represents a path as a Path object instead of a string. The / operator joins parts with the right separator, and the object answers questions about itself:

from pathlib import Path
 
root = Path("logs_demo")
log = root / "app" / "errors.log"
print(log)
print("name:", log.name, "| stem:", log.stem, "| suffix:", log.suffix, "| parent:", log.parent)
print("exists:", log.exists(), "| is_file:", log.is_file(), "| root is_dir:", root.is_dir())
print("absolute:", log.resolve())
 
print("glob('*.log'): ", sorted([p.as_posix() for p in root.glob("*.log")]))
print("rglob('*.log'):", sorted([p.as_posix() for p in root.rglob("*.log")]))

.name is the last part, .stem the name without its suffix, .suffix the extension with its dot, and .parent another Path for the directory above. exists(), is_file() and is_dir() ask the filesystem, the same questions as [[ -e ]], [[ -f ]] and [[ -d ]]. resolve() turns a relative path into an absolute one; you will need it for untrusted paths at the end of the lesson.

glob("*.log") matched nothing, because it only looks in logs_demo itself and every log is one level down. rglob() repeats the same pattern in every subdirectory. An empty list from glob() usually means the files are deeper, not missing. Both return paths in the order the filesystem lists them, so the script sorts them; as_posix() gives the path as a string with forward slashes.

Reading text: encoding and errors

A file holds bytes. To give you text, Python must decode them with an encoding, and if you do not name one it uses the locale's. Name it, as count_errors.py does. It also reads the file line by line: iterating over an open file keeps one line in memory at a time, so a 10 GB log costs no more memory than a small one. read_text() loads the whole file into one string, which is fine for a config file and wrong for a log that can grow.

import sys
from pathlib import Path
 
 
def count_errors(log: Path) -> int:
    hits = 0
    # Iterating over the open file reads one line at a time, whatever its size.
    with open(log, encoding="utf-8") as lines:
        for line in lines:
            if "ERROR" in line:
                hits += 1
    return hits
 
 
for log in sorted(Path(sys.argv[1]).rglob("*.log")):
    print(f"{count_errors(log):>3}  {log.as_posix()}")

The first log went through, then legacy.log stopped the program. The last line of the traceback says why: 0xe9 at position 13 is not valid UTF-8 (od -c prints that byte as the octal value 351). Strict decoding is the default, and for a config file it is what you want: an error, not a guess. A log that is sometimes written by something else must not stop the whole report, so you tell Python what to do with bytes it cannot decode:

With errors="replace" every log was counted, including the ERROR line in legacy.log. The choice of error handler decides what your output says about the bad bytes:

LOG = "logs_demo/app/legacy.log"
 
for errors in ("replace", "backslashreplace"):
    with open(LOG, encoding="utf-8", errors=errors) as f:
        print(f"utf-8, {errors:<17}", f.readline(), end="")
 
with open(LOG, encoding="latin-1") as f:
    print(f"{'latin-1':<24}", f.readline(), end="")

replace puts the replacement character � in place of the byte, so the original value is lost. backslashreplace keeps it visible as \xe9, which is better for evidence you may need to look at later. If you know the file's real encoding, name it: latin-1 decodes the line correctly. For counting and searching, replace is enough; never use an error handler to make a config file "work".

Python can warn you about every open() that leaves the encoding to the locale. Run your scripts once with -X warn_default_encoding:

In Python 3.14 the default comes from the locale, so the same script can decode a file differently on a machine with another locale. PEP 686 makes UTF-8 the default from Python 3.15; passing encoding= works the same on every version.

Writing safely: a temporary file and os.replace

A nightly job rewrites report.txt, and a dashboard reads it. The report lines come from a small module that both versions of the job import:

from pathlib import Path
 
 
def report_line(root: Path, log: Path) -> str:
    """One report line: the log's path under root, its line count and its size."""
    count = 0
    with open(log, encoding="utf-8", errors="replace") as lines:
        for _ in lines:  # _ is the usual name for a value you do not use
            count += 1
    name = log.relative_to(root).as_posix()
    return f"{name:<18} {count:>5} {log.stat().st_size:>6}\n"

import sys
from pathlib import Path
 
from loglines import report_line
 
root, target = Path(sys.argv[1]), Path(sys.argv[2])
# open(..., "w") empties the old report at once, before anything new is written
with open(target, "w", encoding="utf-8") as out:
    for log in sorted(root.rglob("*.log")):
        out.write(report_line(root, log))

Run it once, then make one log unreadable, as a log owned by root would be, and run it again:

The PermissionError stopped the job, and the damage was done before it: open(target, "w") emptied the old report the moment it opened it. What is left holds two of four lines, and the dashboard shows it as if it were complete. A reader that opens the file while the job is still writing sees the same half-written state even when nothing fails.

The fix is to write the new version under another name in the same directory and swap it into place when it is complete:

import os
import sys
import tempfile
from pathlib import Path
 
from loglines import report_line
 
 
def write_report(root: Path, target: Path) -> None:
    """Write the report to a temporary file, then swap it into place in one step."""
    # Same directory as target: os.replace works only within one filesystem.
    # mkstemp creates the file with mode 0600 under a name nobody can predict.
    fd, tmp_name = tempfile.mkstemp(dir=target.parent, prefix=f".{target.name}.")
    try:
        with open(fd, "w", encoding="utf-8") as out:
            for log in sorted(root.rglob("*.log")):
                out.write(report_line(root, log))
            out.flush()
            os.fsync(out.fileno())  # the data is on disk before the rename
        os.replace(tmp_name, target)
    except BaseException:  # also Ctrl-C (SIGTERM needs a handler: scripting-adv)
        os.unlink(tmp_name)
        raise
 
 
def main(argv: list[str]) -> int:
    if len(argv) != 2:
        print("usage: report.py LOGDIR REPORT", file=sys.stderr)
        return 2
    root, target = Path(argv[0]), Path(argv[1])
    try:
        write_report(root, target)
    except OSError as err:
        print(f"report: {err}", file=sys.stderr)
        return 1
    print(f"report: wrote {target}")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

tempfile.mkstemp() creates a new, empty file with an unpredictable name, readable and writable only by you, and returns an open file descriptor (the number the kernel uses for an open file) and the file's name. open(fd, "w", ...) wraps that descriptor in a normal Python file object. os.fsync() asks the kernel to put the data on disk before the rename, so a power cut cannot leave an empty report behind. os.replace() then renames the temporary file over the target in one step: a reader sees the old report or the new one, never a mix. A rename only works within one filesystem, which is why the temporary file is created in the target's directory and not in /tmp. If anything fails, the except block deletes the temporary file and re-raises the error. It catches BaseException so that Ctrl-C (KeyboardInterrupt, which is not an Exception) cleans up too. SIGTERM, which timeout and systemctl stop send, is different: by default it ends Python without running any except or finally block, so the temporary file stays behind; scripting-adv adds the signal handler.

The second run failed with a one-line message and exit status 1, and report.txt still holds the last good report. ls -A shows no leftover .report.txt.* file; the other names are the lesson's scripts, and __pycache__ is where Python caches the compiled loglines module. This is the Python version of the write-then-mv pattern from the Bash course. Here is what /tmp would have done:

/tmp is a tmpfs in memory and the home directory is on ext4. os.replace() asks the kernel for a rename, the kernel refuses to rename across filesystems, and Python raises OSError: [Errno 18] Invalid cross-device link. A report job with its temporary file in /tmp would fail on every run.

File modes: what a new file really gets

stat printed 600 for the new report: owner read and write, nobody else. That came from mkstemp(), and os.replace() keeps the temporary file's mode. Other ways of creating a file give different results, and the umask (the permission bits your shell removes from every new file) is part of the answer:

import os
from pathlib import Path
 
Path("plain.txt").write_text("report\n", encoding="utf-8")  # asks for 0o666
Path("touched.txt").touch(mode=0o600)
# O_EXCL: create a new file or fail; never open one that is already there
fd = os.open("private.txt", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
os.close(fd)
 
for name in ("plain.txt", "touched.txt", "private.txt"):
    print(f"{name:<12} {os.stat(name).st_mode & 0o777:o}")

The numbers after 0o are octal, the same notation as chmod 600. Three operators need a word: | between the os.O_* flags combines them into one number (each flag is one bit), & 0o777 keeps only the permission bits of st_mode, and :o in the f-string prints a number in octal. write_text() asks for 0o666, and the login umask 0002 (Ubuntu's default for a user with a private group) removes write for others: 664, readable by everyone on the host. A report that holds host names, findings or tokens should not be. touch(mode=0o600) and os.open(..., 0o600) ask for owner-only, and a umask can only remove bits, so they get 600. The mode applies only when a file is created: an existing file keeps its old mode whatever you pass.

The second run failed with FileExistsError. O_EXCL means "create a new file or fail", and it also fails when the name is a symbolic link someone planted there. Use it when a file must not already exist, such as a lock or a fresh output file. If you need a mode other than 0600 after mkstemp(), set it on the open descriptor with os.fchmod(fd, 0o640) before the rename.

Untrusted paths: resolve, then check

A tool serves files from uploads/ by name, and the name comes from a request. The tempting check tests the text of the path; the correct one tests where the path leads:

import sys
from pathlib import Path
 
BASE = Path("uploads").resolve()
 
 
def naive_path(name: str) -> Path:
    # WRONG: checks the text of the path, not the place it leads to
    candidate = BASE / name
    if not candidate.is_relative_to(BASE):
        raise ValueError(f"outside uploads/: {name!r}")
    return candidate
 
 
def safe_path(name: str) -> Path:
    candidate = (BASE / name).resolve()  # removes .. and follows symlinks
    if not candidate.is_relative_to(BASE):
        raise ValueError(f"outside uploads/: {name!r}")
    return candidate
 
 
for check in (naive_path, safe_path):
    print(check.__name__)
    for name in sys.argv[1:]:
        try:
            print("  OK  ", check(name))
        except ValueError as err:
            print("  DENY", err)

is_relative_to(BASE) is true when a path lies under BASE, but it only compares the parts of the path as text; it does not look at the disk. for check in (naive_path, safe_path) loops over the two functions themselves, and check.__name__ is each function's name.

Both functions accept the two real files. /etc/passwd is refused by both, because joining an absolute path with / discards the base: BASE / "/etc/passwd" is just /etc/passwd. The difference is the .. request. naive_path accepted uploads/../../../../etc/passwd, because as text it starts with the base, but the kernel follows each .. upward and ends at /etc/passwd. safe_path resolved it to /etc/passwd first and refused it.

.. is not the only way out. Someone who can write into uploads/ can leave a symbolic link there:

The name latest/passwd contains no .., and a text filter would pass it. resolve() follows the link to /etc/passwd, and the containment check refuses it.

One gap remains. Between resolve() and open() there is a moment in which someone who can write into uploads/ could swap a file for a link. When the directory is writable by people you do not trust, open with O_NOFOLLOW or relative to a directory descriptor; the scripting-adv lesson on hostile input shows how.

Try this

Write save_note.py NAME TEXT in ~/ps-files. It stores TEXT in notes/NAME with the temporary-file-and-os.replace pattern, the new file must end up with mode 600, and NAME must stay inside notes/ after resolve(). Print saved notes/NAME and exit 0 on success; print one line to stderr and exit 2 for a refused name or a write error. Verify it: after mkdir notes, saving todo.txt prints saved notes/todo.txt and stat -c "%a" notes/todo.txt prints 600; ../escape.txt is refused with exit 2; after ln -s /tmp notes/tmp, tmp/planted.txt is refused too. Then break it: remove .resolve() from the containment check, save ../escape.txt again and confirm that escape.txt now appears next to notes/, outside the directory you meant to allow.

Takeaway

Name the encoding and the error policy on every open(), write files that matter to a temporary file in the same directory and os.replace() them into place, and check the mode the file really got. Resolve any path that comes from outside, then check that it is still under your base directory.

Related