Input Validation, Files and Secrets

The rule

Anything that did not come from your own code is untrusted: command line arguments, environment variables, files, network responses, form fields, file names, and anything a user typed. Validate at the boundary, once, and work with clean values afterwards.

Validate, do not sanitise

# Weak: trying to remove the bad parts
def clean_username(value):
    return value.replace("<", "").replace(">", "").replace("'", "")


# Strong: state what is acceptable and reject everything else
import re

USERNAME = re.compile(r"[a-z][a-z0-9_]{2,19}")


def validate_username(value):
    if not isinstance(value, str):
        raise TypeError("username must be text")
    value = value.strip().lower()
    if not USERNAME.fullmatch(value):
        raise ValueError(
            "username must be 3 to 20 characters, start with a letter, "
            "and contain only letters, digits and underscores"
        )
    return value


for candidate in ["meera_n", "ab", "Meera Nair", "<script>"]:
    try:
        print(validate_username(candidate))
    except (TypeError, ValueError) as error:
        print(f"{candidate!r}: {error}")

A list of allowed shapes can be reasoned about. A list of forbidden characters can always be worked around, because you cannot enumerate every bad case.

Numbers and ranges

def validate_quantity(raw, minimum=1, maximum=1000):
    try:
        value = int(raw)
    except (TypeError, ValueError):
        raise ValueError(f"quantity must be a whole number, got {raw!r}") from None
    if not minimum <= value <= maximum:
        raise ValueError(f"quantity must be between {minimum} and {maximum}")
    return value


for raw in ["5", "0", "9999", "abc", None]:
    try:
        print(validate_quantity(raw))
    except ValueError as error:
        print(f"{raw!r}: {error}")

Note the length check and the range check. Without an upper bound, int("9" * 1_000_000) is a valid integer that takes a very long time to parse and to work with - a denial of service with no exploit required.

MAX_INPUT_LENGTH = 10_000


def read_bounded(handle, limit=MAX_INPUT_LENGTH):
    data = handle.read(limit + 1)
    if len(data) > limit:
        raise ValueError(f"input exceeds {limit} characters")
    return data

Path traversal

from pathlib import Path

UPLOADS = Path("/srv/app/uploads").resolve()


# Vulnerable: the name can escape the directory
def read_unsafe(name):
    return (UPLOADS / name).read_text(encoding="utf-8")


# read_unsafe("../../etc/passwd")     # reads a file far outside uploads


# Safe: resolve, then confirm the result is still inside
def read_safe(name):
    candidate = (UPLOADS / name).resolve()
    if not candidate.is_relative_to(UPLOADS):        # Python 3.9+
        raise ValueError(f"path escapes the upload directory: {name!r}")
    if not candidate.is_file():
        raise FileNotFoundError(name)
    return candidate.read_text(encoding="utf-8")


for name in ["notes.txt", "../../etc/passwd", "sub/../notes.txt"]:
    try:
        read_safe(name)
    except (ValueError, FileNotFoundError) as error:
        print(f"{name!r}: {type(error).__name__}: {error}")

import re
from pathlib import Path

SAFE_NAME = re.compile(r"[\w.-]{1,255}")


def safe_filename(name):
    """Reduce an arbitrary name to something safe to use as a file name."""
    name = Path(name).name                       # strip any directory part
    if name in ("", ".", "..") or not SAFE_NAME.fullmatch(name):
        raise ValueError(f"unsafe file name: {name!r}")
    return name


for name in ["report.txt", "../secret", "a/b.txt", "", "..", "ok-file_1.md"]:
    try:
        print(f"{name!r:<18}-> {safe_filename(name)}")
    except ValueError as error:
        print(f"{name!r:<18}-> rejected: {error}")

Two defences, both needed: reduce the name to its final component with Path(name).name, and confirm the resolved path is still inside the intended directory. resolve() is essential because it collapses .. and follows symbolic links before the check.

Safe file handling

import os
from pathlib import Path

# Refuse to overwrite
try:
    with open("report.txt", "x", encoding="utf-8") as handle:
        handle.write("fresh\n")
except FileExistsError:
    print("refusing to overwrite an existing file")


# Restrictive permissions on a new file holding secrets
path = Path("credentials.txt")
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
    handle.write("token\n")

print(oct(path.stat().st_mode)[-3:])       # 600 - owner only


# Bound how much you read from an untrusted file
MAX_BYTES = 5 * 1024 * 1024


def read_limited(path):
    path = Path(path)
    if path.stat().st_size > MAX_BYTES:
        raise ValueError("file is too large")
    return path.read_text(encoding="utf-8", errors="replace")

import tempfile
from pathlib import Path

# Safe: created with restrictive permissions and a name nobody can predict
with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8",
                                 delete=False, suffix=".tmp") as handle:
    handle.write("working data")
    temporary = Path(handle.name)

print(temporary.name)
temporary.unlink()

# Unsafe: a predictable name in a shared directory invites a race
# path = Path("/tmp/myapp-scratch.txt")

Secrets

# Never do this
API_KEY = "sk_live_51H8xY2KZ..."          # committed to version control forever
DATABASE_PASSWORD = "hunter2"

import os
import sys


def require_secret(name):
    value = os.environ.get(name)
    if not value:
        sys.exit(f"fatal: the environment variable {name} is not set")
    return value


API_KEY = require_secret("API_KEY")
DATABASE_URL = require_secret("DATABASE_URL")

$ export API_KEY="..."          set it in the environment, not in the code
$ echo ".env" >> .gitignore     never commit the file that holds it

Keeping secrets out of logs

import logging
from dataclasses import dataclass, field

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class Credentials:
    username: str
    password: str = field(repr=False)          # excluded from repr

    def __str__(self):
        return f"Credentials({self.username}, ********)"


creds = Credentials("meera", "hunter2")
print(creds)
logger.info("authenticating %s", creds)        # the password never appears


SENSITIVE = {"password", "token", "api_key", "secret", "authorization"}


def safe_for_logging(data):
    return {
        k: ("********" if k.lower() in SENSITIVE else v)
        for k, v in data.items()
    }


logger.info("request: %s", safe_for_logging(
    {"user": "meera", "password": "hunter2", "action": "login"}))

Randomness

import random
import secrets

# Predictable: fine for simulations and games, never for security
print(random.randint(100000, 999999))
print("".join(random.choices("abcdef0123456789", k=32)))

# Unpredictable: use these for anything an attacker should not guess
print(secrets.token_hex(32))               # a session token
print(secrets.token_urlsafe(32))           # safe in a URL
print(secrets.randbelow(1_000_000))        # a one time code
print(secrets.choice(["a", "b", "c"]))

import secrets
import string


def generate_password(length=20):
    alphabet = string.ascii_letters + string.digits + "!@#$%^&*-_"
    while True:
        password = "".join(secrets.choice(alphabet) for _ in range(length))
        if (any(c.islower() for c in password)
                and any(c.isupper() for c in password)
                and any(c.isdigit() for c in password)):
            return password


print(generate_password())

Comparing secrets

import hmac
import secrets

expected = secrets.token_hex(16)
supplied = expected

# Vulnerable: == returns as soon as the first differing byte is found,
# so the time taken leaks how much of the value was correct.
print(supplied == expected)

# Safe: constant time regardless of where the difference is
print(hmac.compare_digest(supplied, expected))

Password storage

import hashlib
import hmac
import secrets

# WRONG: fast hashes are designed to be fast, which helps an attacker
# stored = hashlib.sha256(password.encode()).hexdigest()
# stored = hashlib.md5(password.encode()).hexdigest()


def hash_password(password, iterations=600_000):
    """Store a password using a deliberately slow key derivation function."""
    salt = secrets.token_bytes(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"),
                                 salt, iterations)
    return f"pbkdf2_sha256${iterations}${salt.hex()}${digest.hex()}"


def verify_password(password, stored):
    algorithm, iterations, salt_hex, digest_hex = stored.split("$")
    digest = hashlib.pbkdf2_hmac(
        "sha256", password.encode("utf-8"),
        bytes.fromhex(salt_hex), int(iterations),
    )
    return hmac.compare_digest(digest.hex(), digest_hex)


stored = hash_password("correct horse battery staple")
print(stored[:40], "...")
print(verify_password("correct horse battery staple", stored))     # True
print(verify_password("wrong password", stored))                   # False

• A salt is random per password, so identical passwords produce different stored values and precomputed tables are useless.
• Many iterations make each guess expensive, which matters only to an attacker making millions of them.
• A plain sha256 or md5 of a password is a serious flaw, not a minor one.
• For production, a dedicated password hashing algorithm is better still; pbkdf2_hmac is what the standard library provides.

Failing safely

import logging

logger = logging.getLogger(__name__)


# Leaks internal detail to the user
def handler_unsafe(user_id):
    try:
        return load_user(user_id)
    except Exception as error:
        return f"Error: {error}"          # may expose paths, queries, versions


# Log the detail, return something generic
def handler_safe(user_id):
    try:
        return load_user(user_id)
    except FileNotFoundError:
        logger.warning("user %s not found", user_id)
        return {"error": "not found"}
    except Exception:
        logger.exception("unexpected failure loading user %s", user_id)
        return {"error": "an internal error occurred"}

# Fail closed, not open
def has_access_unsafe(user, resource):
    try:
        return check_permissions(user, resource)
    except Exception:
        return True                       # a failure grants access


def has_access_safe(user, resource):
    try:
        return check_permissions(user, resource)
    except Exception:
        logger.exception("permission check failed")
        return False                      # a failure denies access

Two rules: an error message to a user should never contain a file path, a query, a stack trace or a version number; and when a security check itself fails, the answer is always "denied".

Dependencies and the standard library

import sys

# Anything on sys.path can shadow a standard library module
print(sys.path[0])          # the script's own directory comes FIRST

# A file named json.py or random.py in your project silently replaces
# the real module for every import in the program.

• Never name a file after a module you import.
• Keep PYTHONPATH free of directories other users can write to.
• Review anything you install; installing a package runs its code.
• Prefer the standard library, which is exactly what this path teaches.

A validation layer

import re
from pathlib import Path


class ValidationError(Exception):
    """Raised when submitted data fails validation."""


class Validator:
    """Validate a submitted record against declared rules."""

    PATTERNS = {
        "username": re.compile(r"[a-z][a-z0-9_]{2,19}"),
        "email": re.compile(r"[^@\s]{1,64}@[^@\s]{1,255}\.[a-z]{2,}"),
        "pin": re.compile(r"\d{6}"),
    }

    def __init__(self, uploads):
        self.uploads = Path(uploads).resolve()

    def text(self, value, field, maximum=255):
        if not isinstance(value, str):
            raise ValidationError(f"{field} must be text")
        value = value.strip()
        if not value:
            raise ValidationError(f"{field} is required")
        if len(value) > maximum:
            raise ValidationError(f"{field} may not exceed {maximum} characters")
        return value

    def pattern(self, value, field):
        value = self.text(value, field)
        if not self.PATTERNS[field].fullmatch(value):
            raise ValidationError(f"{field} is not in the expected format")
        return value

    def number(self, value, field, minimum=None, maximum=None):
        try:
            number = int(str(value)[:20])
        except ValueError:
            raise ValidationError(f"{field} must be a whole number") from None
        if minimum is not None and number < minimum:
            raise ValidationError(f"{field} must be at least {minimum}")
        if maximum is not None and number > maximum:
            raise ValidationError(f"{field} must be at most {maximum}")
        return number

    def choice(self, value, field, allowed):
        if value not in allowed:
            raise ValidationError(
                f"{field} must be one of {', '.join(sorted(allowed))}")
        return value

    def upload_path(self, name):
        candidate = (self.uploads / Path(str(name)).name).resolve()
        if not candidate.is_relative_to(self.uploads):
            raise ValidationError("invalid file name")
        return candidate


def validate_submission(data, validator):
    problems = {}
    clean = {}

    checks = {
        "username": lambda v: validator.pattern(v, "username"),
        "email": lambda v: validator.pattern(v, "email"),
        "age": lambda v: validator.number(v, "age", 13, 120),
        "role": lambda v: validator.choice(v, "role", {"viewer", "editor"}),
    }

    for field, check in checks.items():
        try:
            clean[field] = check(data.get(field))
        except ValidationError as error:
            problems[field] = str(error)

    if problems:
        raise ValidationError(problems)
    return clean


validator = Validator(".")

try:
    print(validate_submission({
        "username": "meera_n", "email": "m@example.com",
        "age": "27", "role": "editor",
    }, validator))

    validate_submission({
        "username": "M", "email": "nope", "age": "200", "role": "admin",
    }, validator)
except ValidationError as error:
    for field, message in error.args[0].items():
        print(f"{field:<10}{message}")

Common mistakes

• Filtering out bad characters instead of accepting only good ones.
• Joining a user supplied name onto a directory without resolving and checking it.
• Storing a secret in source code or passing it as a command line argument.
• Using random for tokens, passwords or one time codes.
• Comparing secrets with ==.
• Hashing passwords with a single fast hash and no salt.
• Returning the exception message to the user.
• Failing open when a permission check errors.

Best practices

• Validate at the boundary, with an allow list, and bound every length.
• Resolve every path and confirm it is inside the directory you intended.
• Read secrets from the environment; never commit them.
• Use secrets for anything unguessable and hmac.compare_digest to compare.
• Hash passwords with a salt and many iterations.
• Log the detail, return a generic message, and fail closed.

Practice

1. Write validators for a username, an email and an age, each rejecting three bad inputs.
2. Write a function that safely resolves a user supplied file name inside one directory.
3. Move a hard coded key into an environment variable with a clear failure message.
4. Implement password hashing and verification with a salt, and show two identical passwords producing different stored values.
5. Take a function that returns raw exception text and rewrite it to log the detail and return a generic message.

Conclusion

Validate with an allow list, bound every length, resolve every path and check it, keep secrets in the environment, use secrets for randomness and compare_digest for comparison, and salt and stretch every password. None of it is clever; all of it is necessary.