DEV Community: Ajit Kumar

Moving a production API to Hetzner, part 3: migrating data and cutting over

Ajit Kumar — Fri, 30 Oct 2026 07:21:44 +0000

The new server is hardened and running Docker (part 2).
Now we move the application. The goal is a few minutes of downtime, zero lost
data, and a rollback plan we never have to improvise.

To keep the commands concrete, this part uses a small example application. Swap
in your own names; the method doesn't change.

0. The example application

notes-api is a small REST API: users sign in and create, read and delete
notes. It's the same shape as most small production services:

Piece	What it does	State it holds
`server`	the API, a container image built by CI and pushed to `ghcr.io/<you>/notes-api`	none
`postgres`	PostgreSQL 16: users, notes, schema migrations	everything that matters
`valkey`	Valkey (the open-source Redis fork): response cache and rate-limit counters	short-lived only
`caddy`	reverse proxy; gets and renews the TLS certificate automatically	a certificate it can re-fetch

Users sign in with a token signed by a secret in .env.prod (JWT_SECRET), and
a nightly backup job writes a compressed pg_dump to an S3-compatible bucket.

The repository looks like this:

notes-api/
├── docker/
│   ├── docker-compose.prod.yml
│   └── Caddyfile
├── scripts/
│   └── backup.sh
└── .env.prod.example      # the real .env.prod is never committed

The production compose file, trimmed to what matters for a migration:

# docker/docker-compose.prod.yml
services:
  postgres:
    image: postgres:16-alpine
    container_name: notes_postgres
    environment:
      POSTGRES_USER: notes
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: notes
    volumes: [pgdata:/var/lib/postgresql/data]
    healthcheck: { test: ["CMD", "pg_isready", "-U", "notes"], interval: 5s }
    restart: unless-stopped

  valkey:
    image: valkey/valkey:9-alpine
    container_name: notes_valkey
    command: valkey-server --requirepass ${VALKEY_PASSWORD}
    restart: unless-stopped

  server:
    image: ${APP_IMAGE:-ghcr.io/<you>/notes-api:latest}
    container_name: notes_server
    env_file: ../.env.prod
    depends_on:
      postgres: { condition: service_healthy }
    restart: unless-stopped

  caddy:
    image: caddy:2-alpine
    container_name: notes_caddy
    ports: ["80:80", "443:443", "443:443/udp"]   # the ONLY published ports
    volumes:
      - ./Caddyfile:/etc/caddy/Caddyfile:ro
      - caddy_data:/data
    restart: unless-stopped

volumes: { pgdata: {}, caddy_data: {} }

# docker/Caddyfile
api.example.com {
    reverse_proxy server:8080
}

Three details in there matter later:

Only Caddy publishes ports. Postgres and Valkey are reachable only on Docker's internal network.
image: ${APP_IMAGE:-…:latest}: the image comes from a variable with a default, so a deploy can pin an exact build without editing the file.
The API runs its database migrations on startup. That decides the order in which we start things.

Placeholders, as in parts 1 and 2: the old server is old-box (203.0.113.10),
the new one is new-box (198.51.100.20), and the API is api.example.com.

1. Decide what moves, and what doesn't

The most useful migration question is what state lives where.

State	Where	Moves?	Why
Users, notes, migration history	Postgres volume	yes	the only durable state on the box
Secrets (`.env.prod`)	a file next to the compose file	yes, byte-identical	DB and Valkey passwords, `JWT_SECRET`, storage and API keys
Cache, rate-limit counters	Valkey	no	rebuilt on demand; counters resetting to zero is harmless
TLS certificate	Caddy's volume	no	Caddy fetches a fresh one in seconds
Backups	S3-compatible bucket	no	already off the box; the new server writes to the same bucket
Code	container image in a registry	pulled	never built on the server (section 3)

Not moving Valkey is worth a moment's thought, because it's tempting to copy
everything "to be safe". Ask of each kind of key: what happens if it's gone?
Cache entries are refetched from Postgres. Rate-limit counters start again from
zero. Nobody is signed out, because sign-in tokens are verified with
JWT_SECRET, which travels in .env.prod, not in Valkey. If your app keeps
sessions in Redis or Valkey, that changes the answer: either copy it or accept
that everyone signs in again.

2. Copying between two servers through the laptop

The trick

The old and new servers can't log in to each other, and they shouldn't be able
to: that would mean a private key for one sitting on the other. But the laptop can
reach both (part 2 set up old-box and new-box in ~/.ssh/config). So we
connect the two SSH sessions with a pipe. First clone the repository on the
new box, then copy the secrets file across:

ssh new-box 'git clone https://github.com/<you>/notes-api ~/notes-api'
ssh old-box 'cat ~/notes-api/.env.prod' | ssh new-box 'umask 077 && cat > ~/notes-api/.env.prod'

Reading the second line left to right:

ssh old-box 'cat …' runs cat on the old server. Its output, the file, comes back over the first encrypted SSH connection as that command's stdout.
| is a pipe on the laptop. It connects the first command's stdout to the second command's stdin, in memory. Nothing is written to the laptop's disk, and nothing is printed to its terminal.
ssh new-box '… cat > …' runs cat on the new server, whose stdin is the second encrypted connection. It writes the bytes to the destination file.
umask 077 is set first, so the file is created with mode 600 (owner-only). That's different from fixing the permissions afterwards, which leaves a moment when the secrets are world-readable.

It's one command, with both servers' credentials staying on the laptop. You
don't have to log in to the old box, download, log in to the new box, upload,
and then remember to delete the local copy of your production secrets.

Alternatives, and why not:

scp -3 old-box:f new-box:f does the same routing through the laptop. Fine too; the pipe is easier to combine with umask and checks.
Agent forwarding (ssh -A old-box, then scp from there to the new box) works, but while you're connected, anyone with root on the old box can use your agent to authenticate as you anywhere. Avoid it on servers you're about to decommission.

Verify it arrived intact

ssh old-box 'sha256sum ~/notes-api/.env.prod'; ssh new-box 'sha256sum ~/notes-api/.env.prod; stat -c "%a %U" ~/notes-api/.env.prod'

9f2c41d7…b83e0  /home/ubuntu/notes-api/.env.prod
9f2c41d7…b83e0  /home/ubuntu/notes-api/.env.prod
600 ubuntu

A cryptographic hash is a fingerprint: change one byte and it changes
completely. Matching hashes prove the files are identical, and you never had to
display the secrets to compare them. One wrong character in JWT_SECRET would
otherwise sign every user out at cutover.

Copy only .env.prod. Leave behind any development .env or stray .env.prod.bak
on the old box. With no plain .env present, a command that forgets
--env-file .env.prod fails loudly instead of quietly running with development
settings.

3. Pull the image, never build on the box

ssh old-box 'docker inspect notes_server --format "{{.Config.Image}}"'
# ghcr.io/<you>/notes-api:3f9c2ab
ssh new-box 'docker pull ghcr.io/<you>/notes-api:3f9c2ab'

The repository gives us the compose file and scripts. The application itself
comes as a prebuilt image, and the first command asks the old box exactly
which tag it's running. Pull that one, not latest. Two reasons:

Memory. Compiling a release build (Rust, Go with CGO, a big TypeScript app) can need several GB of RAM. On a 4 GB box with Postgres running, that's an out-of-memory kill waiting to happen. CI builds the image; the server only downloads it.
Same bits. Running the old box's exact build means the migration changes one thing at a time: the host. If something breaks after cutover, it's the move, not a code change. Upgrades come afterwards, separately.

4. Start the data services only

ssh new-box 'cd ~/notes-api && docker compose -f docker/docker-compose.prod.yml --env-file .env.prod up -d postgres valkey && docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Ports}}"'

NAMES            STATUS                   PORTS
notes_postgres   Up 2 minutes (healthy)   5432/tcp
notes_valkey     Up 2 minutes             6379/tcp

Read the PORTS column carefully. 5432/tcp means "exposed inside Docker's
network"; 0.0.0.0:5432->5432/tcp would mean "published to the world", which
is the Docker-bypasses-ufw problem from parts 1 and 2.

Two services are deliberately not started yet:

The API server. It runs migrations on startup, so it would create an empty schema, and the restore in section 7 would then collide with it.
Caddy. It would immediately try to get a certificate for api.example.com, which still points at the old server. The challenge would fail, and repeated failures run into Let's Encrypt's rate limits.

5. Rehearse the restore before you need it

A backup you have never restored is a hope, not a backup.

The nightly job on the old box uploads notes-YYYYMMDD.sql.gz to a bucket.
Restoring the newest one into a scratch database on the new box, before
cutover, proves three things at once: the storage credentials work from the new
IP (some providers filter tokens by source address), the restore path works on
this machine, and the backups are actually good.

# list the bucket (the AWS CLI works with any S3-compatible store)
ssh new-box 'set -a; . ~/notes-api/.env.prod; set +a;
  docker run --rm -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY amazon/aws-cli \
    --endpoint-url "$S3_ENDPOINT" s3 ls "s3://$BACKUP_BUCKET/" | tail -3'

2026-10-27 03:02:11      41208 notes-20261027.sql.gz
2026-10-28 03:02:09      41377 notes-20261028.sql.gz
2026-10-29 03:02:14      41503 notes-20261029.sql.gz

set -a; . file; set +a loads .env.prod into the shell as environment
variables without printing them, so the credentials reach the container
through -e NAME and never appear on a command line, where ps could show them.

ssh new-box 'set -a; . ~/notes-api/.env.prod; set +a;
  docker run --rm -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY amazon/aws-cli \
    --endpoint-url "$S3_ENDPOINT" s3 cp "s3://$BACKUP_BUCKET/notes-20261029.sql.gz" - > ~/rehearsal.sql.gz
  docker exec notes_postgres createdb -U notes restore_test
  zcat ~/rehearsal.sql.gz | docker exec -i notes_postgres psql -U notes -d restore_test -q -v ON_ERROR_STOP=1
  docker exec notes_postgres psql -U notes -d restore_test -tAc "select count(*) from users; select count(*) from notes"
  docker exec notes_postgres dropdb -U notes restore_test; rm ~/rehearsal.sql.gz'

57
1342

Download, restore into a throwaway database, count something you know is there,
drop it. ON_ERROR_STOP=1 makes psql stop at the first error instead of
skipping it and carrying on. Non-zero counts mean the backup is real.

A bug a rehearsal like this can turn up

When I ran this rehearsal against my real service, the newest backup in the
bucket was 26 days old. The job had run every night and failed every night,
and no alert had reached anyone. Translated to this example, the log said:

dump ok: notes-20261029.sql.gz (43503 bytes)
gzip: stdout: Broken pipe
ERROR: dump does not contain table notes — wrong database?

The script sanity-checked each dump before uploading it:

set -o pipefail
zcat "$OUT" | grep -q "CREATE TABLE public.notes" || die "dump does not contain table notes"

Here's what happens:

grep -q exits as soon as it finds the first match. That's the point of -q.
zcat is still writing the rest of the file into the pipe. Writing to a pipe whose reader has gone away gets the writer killed with SIGPIPE, hence Broken pipe.
pipefail (part 2: "fail if any stage fails") sees zcat die and fails the whole pipeline.
The script reports "no table notes" because it found it.

It worked for months because it's a race. A Linux pipe buffers 64 KB. While
the decompressed dump was small, zcat had usually written all of it before
grep found its match and quit, so nothing was left to write and no SIGPIPE
came. Once the database grew, zcat still had data to write when grep exited,
and it lost the race every night.

The fix is to let grep read to the end and throw its output away:

zcat "$OUT" | grep "CREATE TABLE public.notes" >/dev/null || die "…"

Lessons: grep -q and head inside a pipefail pipeline are a classic trap;
a check can fail because the thing it checks for is there; and backup
alerting needs its own test. Section 7's checklist triggers the failure alert
on purpose.

6. DNS: lower the TTL a day ahead

How DNS caching works

When a client looks up api.example.com, the answer comes with a TTL (time to
live) in seconds. Every resolver along the way (your ISP's, 1.1.1.1, 8.8.8.8,
the OS cache) may reuse that answer until the TTL expires. With a TTL of 300, after
you change the record, some clients keep going to the old IP for up to 5 minutes.
With a TTL of a day, for up to a day.

So the TTL has to be lowered in advance, by at least the old TTL's
duration, because resolvers that cached the record before the change still hold
it with the old expiry:

The day before, set the api record's TTL to 60 seconds in your DNS provider,
then confirm what resolvers see:

dig api.example.com @1.1.1.1 +noall +answer
# api.example.com.  60  IN  A  203.0.113.10
#                   ^^ remaining TTL in seconds

DNS-only, not proxied

Many DNS providers can also proxy traffic through their CDN (Cloudflare calls it
the "orange cloud"). The API record stays DNS-only: Caddy gets its own
certificate with the ACME HTTP-01 challenge, and that needs Let's Encrypt to
reach this server on port 80 directly:

That's why port 80 is open in both firewalls, and why Caddy only starts after
DNS points here.

7. The cutover

Pick a quiet hour and keep three terminals open: old-box, new-box, and the
laptop. Before starting, check two things on the new box: no reboot is pending
(ls /var/run/reboot-required), and the TTL from section 6 has been 60 seconds
for at least the old TTL's duration.

The sequence is designed so that only one database accepts writes at any
moment:

1. Freeze writes. Stop only the API on the old box. From this moment, never
start it again (section 8 explains why):

ssh old-box 'cd ~/notes-api && docker compose -f docker/docker-compose.prod.yml --env-file .env.prod stop server'

Caddy keeps running and answers 502 Bad Gateway. Clients see an error and retry
later. If you receive webhooks (payments, GitHub, anything), the sender treats a
non-2xx response as a failed delivery and retries with backoff, typically for
hours or days, so no event is lost. A 502 is the cheapest possible maintenance
page.

2. Dump straight into the new box:

ssh old-box 'docker exec notes_postgres pg_dump -U notes -Fc notes' | ssh new-box 'umask 077 && cat > ~/cutover.dump && ls -l ~/cutover.dump'

-Fc is Postgres's custom format: compressed, and readable by pg_restore,
which can restore objects selectively and in dependency order. A plain SQL dump
works too, but this is smaller and more flexible. It's the same laptop-pipe
pattern as section 2. Keep cutover.dump afterwards: it's an exact snapshot of
the moment of the move.

3. Restore and compare:

ssh new-box 'cd ~/notes-api && docker compose -f docker/docker-compose.prod.yml --env-file .env.prod exec -T postgres pg_restore -U notes -d notes --no-owner --exit-on-error < ~/cutover.dump && echo RESTORED'

exec -T: no TTY, so the dump can be fed on stdin.
--no-owner: don't try to recreate the original object owners; everything belongs to the connecting user.
--exit-on-error: stop at the first error rather than restoring half a database and reporting success.

Then run the same query on both boxes and require identical output:

Q="select 'users='||(select count(*) from users)||' notes='||(select count(*) from notes)||' migration='||(select max(version) from schema_migrations)"
ssh old-box "docker exec notes_postgres psql -U notes -d notes -tAc \"$Q\""
ssh new-box "docker exec notes_postgres psql -U notes -d notes -tAc \"$Q\""

users=57 notes=1342 migration=20261012093000
users=57 notes=1342 migration=20261012093000

Count your important tables plus the highest applied migration. The table that
records migrations depends on your framework: schema_migrations in Rails and
golang-migrate, django_migrations, flyway_schema_history,
_sqlx_migrations. Counts don't prove every byte matches, but with writes
frozen, any difference means something went wrong. If they differ, stop and roll
back (section 8).

4. Start the API, pinned to the same build, and check it from inside the
container, before any traffic arrives. Setting APP_IMAGE on the command line
fills the ${APP_IMAGE:-…} variable in the compose file:

ssh new-box 'cd ~/notes-api && APP_IMAGE=ghcr.io/<you>/notes-api:3f9c2ab docker compose -f docker/docker-compose.prod.yml --env-file .env.prod up -d server && sleep 10 && docker exec notes_server wget -qO- http://localhost:8080/health'

{"status":"ok","build":"3f9c2ab","database":"ok","cache":"ok"}

The migrations it runs at startup find the schema already at the latest version
and do nothing. That's exactly what we want.

5. Point DNS at the new IP. In your DNS provider, change the api A record
to 198.51.100.20, DNS-only. Then watch two public resolvers agree:

dig +short api.example.com @1.1.1.1; dig +short api.example.com @8.8.8.8

198.51.100.20
198.51.100.20

6. Start Caddy and wait for the certificate:

ssh new-box 'cd ~/notes-api && APP_IMAGE=ghcr.io/<you>/notes-api:3f9c2ab docker compose -f docker/docker-compose.prod.yml --env-file .env.prod up -d && docker logs -f notes_caddy 2>&1 | grep -m1 -iE "certificate obtained|error"'

{"level":"info","msg":"certificate obtained successfully","identifier":"api.example.com"}

That line ends the downtime. For a database this size, steps 1 to 6 take under
ten minutes.

Verification checklist

Run each check from the laptop, against the public name, so the whole path
(DNS, firewall, Caddy, API, database) is exercised:

curl -s https://api.example.com/health
curl -sI https://api.example.com/health | grep -i strict-transport-security

[ ] /health reports "build":"3f9c2ab": the new box is answering, on the same build.
[ ] The Strict-Transport-Security header is present. In this app it's only sent in production mode, so it proves the right env file is loaded.
[ ] Sign in with an existing account. You don't have to sign in again afterwards, because JWT_SECRET moved intact.
[ ] A write and a read: create a note, fetch it, delete it. This is the first write the new database accepts.

ID=$(curl -s -X POST https://api.example.com/notes -H "Authorization: Bearer $TOKEN" -d '{"text":"migration check"}' | jq -r .id)
curl -s https://api.example.com/notes/$ID -H "Authorization: Bearer $TOKEN"
curl -s -X DELETE https://api.example.com/notes/$ID -H "Authorization: Bearer $TOKEN"

[ ] Webhooks: in each sender's dashboard, deliveries that failed during the freeze should now show retries answered with 200. Replay any that don't.
[ ] Backups move too: install the nightly backup job on the new box, disable it on the old one (two servers writing dumps into one bucket is confusing at best), run it once by hand, and check the new dump appears in the bucket. Then make it fail on purpose (for example with a wrong bucket name) and confirm the alert actually reaches you.

8. Rollback, and the one rule

Rollback is easy until the new box accepts its first write, and dangerous
after:

When	How to roll back
Before DNS changed (counts differ, API won't start…)	`start server` on the old box. Nothing was written anywhere else
After the new box accepted writes	stop the new API, copy its database back (the same pipe, reversed), restore it on the old box, then start the old API and point DNS back

⛔ The rule: once the old API is stopped, never start it again on its own.
Starting it while the new box is live gives you two databases accepting writes,
which is called split-brain. Clients with the old IP still cached write to one
server, everyone else to the other, and there's no automatic merge. Whichever box
you keep, the other one's writes are lost.

9. Decommission, slowly

Keep the old server stopped but intact for two weeks. It's the last full copy
from before the move, and it's your fallback if something only shows up after a
few days. Then:

Take one final dump and archive it.
On the old box: docker compose -f docker/docker-compose.prod.yml down -v (the -v removes the volumes, meaning the data), and shred the env files (shred -u .env.prod).
Rotate every secret that lived on it: storage tokens, API keys, webhook secrets. The disk is going back to a provider you no longer control. The same values now live in the new box's .env.prod, so each rotation means updating that file and restarting the server. Rotating JWT_SECRET signs every user out; do it deliberately, with a heads-up.
Remove the old server's address from any allowlists, and its block from ~/.ssh/config.
Delete the server in the old provider's console, and make sure its backups or snapshots go with it.

What the whole series boils down to

Measure, then size. 4 GB with swap was plenty.
Keep the IP (auto delete off) and firewall outside the VM, because Docker ignores ufw.
Never risk a lockout: two passes, test in between, and negative tests at the end.
Know what state lives where and move only the durable parts.
Copy server-to-server through the laptop with an SSH pipe, and verify with a hash.
Rehearse the restore. It's how you find out whether your backups are real.
Freeze writes, move once, change DNS last, and never let two databases accept writes.

If this helped, or you've hit a different trap moving servers, tell me in the
comments.

Moving a production API to Hetzner, part 2: hardening and configuring the server

Ajit Kumar — Fri, 30 Oct 2026 06:41:47 +0000

In part 1 we created a Hetzner CX23 behind a Cloud Firewall, with a
cloud-init config meant to harden it on first boot. This part does that hardening
by hand, one step at a time, explains every line, and does it without ever
risking a lockout.

(cloud-init is the tool in every cloud image that runs once on first boot and
applies the config you paste into the Cloud config box: users, files,
packages, commands. Part 1, section 5 has the full config and how it works.)

Why by hand, if cloud-init can do it?

You'll know what each setting does, so you can debug it when it misbehaves.
It works anywhere: on a provider without cloud-init, on a server someone else created, or when cloud-init failed or was never given its config.
It doubles as the verification. If you did use cloud-init, skip to section 5 ("Verify like an attacker") and run the checks. They prove the config was applied, not just pasted.

Placeholders, as before: the new server is 198.51.100.20 and the old one is
203.0.113.10.

1. `~/.ssh/config`: two servers, two keys, short commands

The problem it solves

During a migration you constantly talk to two machines, each with its own
address, port and key. Typing
ssh -p 2222 -i ~/.ssh/hetzner_ed25519 ubuntu@198.51.100.20 every time is slow
and error-prone. ~/.ssh/config gives each server a short name:

Host old-box
  HostName 203.0.113.10
  Port 2222
  User ubuntu
  IdentityFile ~/.ssh/oldhost_key.pem
  IdentitiesOnly yes

Host new-box
  HostName 198.51.100.20
  Port 2222
  User ubuntu
  IdentityFile ~/.ssh/hetzner_ed25519
  IdentitiesOnly yes

Now ssh new-box means all of that. So do scp file new-box:, rsync … new-box:,
and git clone new-box:…. Every OpenSSH-based tool reads this file.

You can append a block without opening an editor:

printf '\nHost new-box\n  HostName 198.51.100.20\n  Port 2222\n  User ubuntu\n  IdentityFile ~/.ssh/hetzner_ed25519\n  IdentitiesOnly yes\n' >> ~/.ssh/config

Why `IdentitiesOnly yes`

Without it, the client offers the server every key in your ssh-agent, one by
one, before the one you named. sshd allows only a few attempts per connection
(MaxAuthTries, default 6). With enough keys loaded you get
Too many authentication failures before the right key is ever tried.
IdentitiesOnly yes means "offer only the IdentityFile named here". It also
stops you from accidentally presenting your work key to a personal server.

2. First login, and the plan for not locking yourself out

On a stock Hetzner image, before any hardening, there is exactly one account you
can reach: root, on port 22, with the SSH key you selected at creation.
Hetzner writes that key into /root/.ssh/authorized_keys and sets no root
password. So that's where we start:

ssh -p 22 -i ~/.ssh/hetzner_ed25519 -o IdentitiesOnly=yes root@198.51.100.20 'id; ss -tln'

(If cloud-init did run, root login is already disabled and this is refused. That's
your cue to skip to the checks in section 5.)

What happened to me: I created the server and forgot to paste the cloud
config. My first ssh -p 2222 got Connection refused, because nothing was
listening on 2222 (section 6 explains how to read that). Recreating the server
would have meant giving up a CX23 that was close to sold out, so I hardened it
by hand, as below.

Running a command after the host ('id; ss -tln') executes it remotely and
returns. No interactive shell is needed, which makes every step repeatable and
easy to paste into notes.

The lockout problem

Hardening SSH means changing the very thing you're connected through. Get one
line wrong, for example disabling root before the new user's key works, and the
next connection fails. On a cloud server, your remaining way in is the provider's
web console. Remember from part 1 that Hetzner sets no root password when you
use an SSH key, so even the console needs a password reset first.

The fix is to make changes in two passes, with a test in between, so that at
every moment at least one known-good way in still works:

3. Pass 1, explained block by block

The whole pass is one script, sent from the laptop and run on the server as root.
'bash -s' tells the remote bash to read its script from standard input, and the
heredoc (<<'EOF' … EOF) supplies it. The quotes around 'EOF' stop your
local shell from expanding $variables inside, so they reach the server as
written.

ssh -p 22 -i ~/.ssh/hetzner_ed25519 -o IdentitiesOnly=yes root@198.51.100.20 'bash -s' <<'EOF'
set -euo pipefail
export DEBIAN_FRONTEND=noninteractive

# --- user ---
id ubuntu >/dev/null 2>&1 || adduser --disabled-password --gecos "" ubuntu
usermod -aG sudo ubuntu
echo 'ubuntu ALL=(ALL) NOPASSWD:ALL' > /etc/sudoers.d/90-ubuntu && chmod 440 /etc/sudoers.d/90-ubuntu
install -d -m 700 -o ubuntu -g ubuntu /home/ubuntu/.ssh
install -m 600 -o ubuntu -g ubuntu /root/.ssh/authorized_keys /home/ubuntu/.ssh/authorized_keys

# --- packages ---
apt-get update -q
apt-get -y -q -o Dpkg::Options::=--force-confold upgrade
apt-get -y -q install ufw unattended-upgrades

# --- sshd ---
cat > /etc/ssh/sshd_config.d/10-hardening.conf <<'CONF'
Port 22
Port 2222
PasswordAuthentication no
KbdInteractiveAuthentication no
PermitRootLogin prohibit-password
PubkeyAuthentication yes
CONF
sshd -t
systemctl daemon-reload
systemctl restart ssh.socket || true
systemctl restart ssh || true

# --- docker log rotation (read when Docker is installed later) ---
mkdir -p /etc/docker
echo '{"log-driver": "json-file", "log-opts": {"max-size": "10m", "max-file": "3"}}' > /etc/docker/daemon.json

# --- host firewall ---
ufw default deny incoming; ufw default allow outgoing
for r in 22/tcp 2222/tcp 80/tcp 443/tcp 443/udp; do ufw allow $r; done
ufw --force enable

# --- swap ---
if [ ! -f /swapfile ]; then fallocate -l 2G /swapfile && chmod 600 /swapfile && mkswap /swapfile && swapon /swapfile && echo '/swapfile none swap sw 0 0' >> /etc/fstab; fi

ss -tln | grep -E ":(22|2222) "
EOF

`set -euo pipefail`

These are bash's "fail loudly" settings: -e stops at the first failing command,
-u treats an unset variable as an error, and pipefail makes a pipeline fail
if any stage fails, not just the last. Part 3 shows pipefail biting back,
too.

The user

adduser --disabled-password: the account has no password at all, so it can only be reached with a key. id ubuntu || … makes the line safe to run twice.
sudoers.d/90-ubuntu with NOPASSWD: ubuntu can run sudo without a password, which is what Ubuntu cloud images give their default user. It's a trade-off: if someone gets this account they get root. But a password-less account with a sudo password needs a password set somewhere, and that's a secret to manage. With key-only SSH, the key is the boundary. chmod 440 because sudo refuses sudoers files that others can write to.
install -d -m 700 / install -m 600: create .ssh and copy root's authorized_keys with the right owner and permissions in one step. The permissions aren't cosmetic. sshd's StrictModes (on by default) ignores an authorized_keys file that other users could modify, and the symptom is a baffling Permission denied (publickey).

Packages

DEBIAN_FRONTEND=noninteractive and --force-confold stop apt from opening an
interactive "a config file changed, which version do you want?" prompt in the
middle of a non-interactive SSH session. confold keeps the existing file.

sshd: drop-in files and "first value wins"

Instead of editing /etc/ssh/sshd_config, we add a file in
/etc/ssh/sshd_config.d/. The main config includes that directory near its
top, and the files load in name order. For most settings sshd keeps the
first value it reads, so 10-hardening.conf beats the image's own
50-cloud-init.conf and anything below in the main file. Package upgrades never
touch your drop-in, either.

Line	Meaning
`Port 22`, `Port 2222`	listen on both (repeatable directive)
`PasswordAuthentication no`	no password logins, so bots guessing passwords get nowhere
`KbdInteractiveAuthentication no`	closes the other password path (PAM "keyboard-interactive"), which can otherwise ask for a password even with the line above
`PermitRootLogin prohibit-password`	pass 1 only: root may log in with a key, our safety line. Pass 2 changes it to `no`
`PubkeyAuthentication yes`	keys, explicitly

Two commands make it safe:

sshd -t tests the configuration and exits non-zero on a syntax error. Because of set -e, a typo stops the script before the restart, so the old, working sshd keeps running.
sshd -T (used later) prints the effective configuration after all files are merged. That's the only reliable way to know which value won.

Why `ssh.socket` and not just `ssh`

Since Ubuntu 24.04, sshd is socket-activated: systemd holds the listening
socket (ssh.socket) and starts sshd when a connection arrives. Which ports that
socket listens on is generated from sshd_config by a systemd generator, and
the generator only runs on systemctl daemon-reload. So changing Port takes
three steps: reload (regenerate the socket), restart ssh.socket (rebind the
ports), restart ssh. Skip the reload and sshd's config says 2222 while nothing
listens there. || true keeps the script going on releases that aren't
socket-activated.

The script ends by printing what's listening. We got
0.0.0.0:22, 0.0.0.0:2222, [::]:22 and [::]:2222.

ufw: the host firewall

ufw ("uncomplicated firewall") is a front end for the kernel's netfilter
(nftables). Default deny incoming and allow outgoing, then open exactly
what's needed. 22 is allowed before ufw enable, otherwise enabling it would
cut the very session doing the enabling.

Docker bypasses ufw for published ports (see part 1), so ufw protects sshd and
anything else running directly on the host, and the Cloud Firewall protects
everything. Two layers, each covering the other's blind spot. That's
defence in depth.

Swap

A 4 GB server with no swap handles a memory spike (a big query, a container
starting) by invoking the kernel's OOM killer, which usually kills the biggest
process: Postgres. A 2 GB swapfile turns that crash into a brief slowdown.
fallocate reserves the space instantly, mkswap formats it, swapon activates
it, and the /etc/fstab line brings it back after a reboot.

4. Test, then pass 2

Before taking anything away, prove the new way in works:

ssh new-box 'whoami; sudo -n true && echo sudo-ok'

ubuntu
sudo-ok

sudo -n means "non-interactive": fail instead of prompting. So this proves
both the key and passwordless sudo. Only now, as ubuntu, turn root off:

ssh new-box 'sudo sed -i "s/^PermitRootLogin .*/PermitRootLogin no/" /etc/ssh/sshd_config.d/10-hardening.conf && sudo sshd -t && sudo systemctl restart ssh && sudo sshd -T | grep -E "^(port|permitrootlogin|passwordauthentication) "'

port 22
port 2222
permitrootlogin no
passwordauthentication no

5. Verify like an attacker

Checking the config proves what you intended. Negative tests prove what's
enforced: try what should fail and watch it fail.

ssh -p 2222 -i ~/.ssh/hetzner_ed25519 -o IdentitiesOnly=yes root@198.51.100.20 true
# root@198.51.100.20: Permission denied (publickey).

ssh -p 2222 -o PubkeyAuthentication=no -o PreferredAuthentications=password ubuntu@198.51.100.20 true
# ubuntu@198.51.100.20: Permission denied (publickey).

The second line is the interesting one. We asked for password authentication,
and the server's reply lists only (publickey) as acceptable. It doesn't offer
a password prompt at all.

Then the rest of the configuration in one go:

ssh new-box 'sudo ufw status verbose; ss -tln; timedatectl show -p NTPSynchronized; swapon --show; free -h; df -h /; cat /etc/apt/apt.conf.d/20auto-upgrades; systemctl is-enabled unattended-upgrades; ls /var/run/reboot-required 2>/dev/null && echo REBOOT-NEEDED'

What to look for, and why each matters:

Check	Expect	Why
`ufw status`	active, deny incoming, five allow rules	the host layer is on
`ss -tln`	only `:22`, `:2222` and `127.0.0.53/54:53`	nothing unexpected listening. `127.0.0.53` is the local DNS resolver (systemd-resolved), bound to loopback only
`NTPSynchronized=yes`	yes	clock accuracy is a security property: TLS certificate validity, TOTP codes (30-second windows) and signed webhooks (Paddle allows about 5 minutes of drift) all fail on a wrong clock
swap / RAM / disk	2 GB swap, ~3.7 GB RAM, ~32 GB free	sanity
`20auto-upgrades`	both lines `"1"`	daily package-list refresh and daily unattended security upgrades
`reboot-required`	present after the big upgrade	a new kernel or libc is installed but not running yet

Automatic updates, and the reboot they don't do

unattended-upgrades installs security updates every day. During pass 1,
needrestart restarted the services whose libraries changed and listed the ones
it deferred. It doesn't reboot: a new kernel only takes effect after a
reboot, and the system flags this with /var/run/reboot-required. Automatic
reboots stay off (the default), because a reboot in the middle of a
database write is worse than a few days on the old kernel.

The best time to reboot is now, while the box is empty. It's also a free test
that everything comes back on its own:

ssh new-box 'sudo reboot'
# wait ~30–60 s
ssh new-box 'whoami; uptime; ss -tln | grep -E ":(22|2222) "; sudo ufw status | head -1; swapon --show --noheadings; ls /var/run/reboot-required 2>/dev/null || echo no-reboot-needed'

Both SSH ports, ufw and swap all survived.

6. Reading a port test: open, refused, filtered

From the laptop, without installing nmap:

for p in 22 2222 443 5432; do timeout 5 bash -c "</dev/tcp/198.51.100.20/$p" 2>/dev/null && echo "$p open" || echo "$p closed/filtered"; done

/dev/tcp/HOST/PORT isn't a real file. It's a bash feature: redirecting from it
opens a TCP connection. timeout 5 caps the wait.

There are three possible answers, and telling them apart is the most useful
network-debugging skill I know:

Result	Meaning	What to look at
open	something accepted the connection	—
refused (instant)	reached the machine, nothing listening	the service: is it running, on that port?
timeout	dropped on the way	firewalls: cloud, host, or your own network

Two real examples from this migration:

ssh -p 2222 → Connection refused on the first try. The Cloud Firewall was fine. sshd simply wasn't on 2222 yet, because cloud-init hadn't run.
Port 22 → timeout from home, even though sshd listened on 22, ufw allowed it and the Cloud Firewall allowed it. Every layer on the server said yes. That points at the path: some home routers and ISPs drop outbound port 22. The way to prove it is to test from a third machine on a different network, such as the old server (ssh old-box 'timeout 5 bash -c "</dev/tcp/198.51.100.20/22" && echo open') or a phone hotspot. This is exactly why SSH listens on two ports.

The final scan read 2222 open, 443 closed/filtered (nothing serves it
yet), and 5432 closed/filtered. The database port is invisible from outside,
which is what we want.

7. Docker

ssh new-box 'curl -fsSL https://get.docker.com | sudo sh && sudo usermod -aG docker ubuntu'

get.docker.com is Docker's official convenience script. It adds Docker's own apt repository and installs the engine and the Compose plugin. Piping a script into sh is fine when you trust the source and it comes over HTTPS; read it first if you want (curl -fsSL https://get.docker.com | less). If it doesn't yet support a brand-new Ubuntu release, add the apt repository by hand from Docker's docs instead.
usermod -aG docker ubuntu lets ubuntu run docker without sudo. ⚠️ The docker group is root-equivalent: anyone in it can start a container that mounts / and edit any file. Here ubuntu already has passwordless sudo, so nothing changes, but never add a less-trusted user to it.

Group membership is read at login, so open a new connection to check:

ssh new-box 'docker --version && docker compose version && docker info --format "{{.LoggingDriver}}"'

Docker version 29.x
Docker Compose version v5.x
json-file

json-file confirms Docker read our daemon.json from pass 1. That file caps
each container's log at 3 × 10 MB. Without it Docker never rotates container
logs, and on a 40 GB disk the reverse proxy's access log eventually fills the
disk. Disk-full is a classic way for a small server to die in month five.

Where we are

Layer	State
Perimeter	Cloud Firewall: 22, 2222, 80, 443/tcp+udp, ICMP
Host	ufw default-deny with the same ports
SSH	key-only, root disabled, on 22 and 2222; verified by negative tests
OS	fully patched, rebooted, automatic security updates on, clock synced
Memory	4 GB + 2 GB swap
Runtime	Docker with log rotation

In part 3 we move the application: copying a secrets file from server to
server through the laptop without it ever touching the laptop's disk, pulling
the image instead of building it, rehearsing a restore from backups (which turned
up a real bug), and planning a cutover with a few minutes of downtime and a clear
rollback.

Moving a production API to Hetzner, part 1: choosing and creating the server

Ajit Kumar — Fri, 30 Oct 2026 05:49:11 +0000

I recently moved a small production API from one VPS provider to Hetzner Cloud.
The stack is ordinary: a Rust web server, PostgreSQL, Valkey (the open-source
Redis fork), and Caddy as the TLS-terminating reverse proxy, all under Docker
Compose. Encrypted blobs live in S3-compatible object storage, which doesn't
move at all.

These posts are both the record of what I did and a tutorial. Each step comes
with the command, what it does, and why it works that way, because
most migration guides give the commands and skip the reasons, and the reasons
are what you need when something goes wrong. Plenty did go wrong, and I've left
those parts in.

Throughout, the addresses are placeholders from the documentation ranges reserved for examples (RFC 5737 and RFC 3849):

Placeholder	Means
`203.0.113.10`	the old server
`198.51.100.20`	the new Hetzner server
`2001:db8:1234:5678::/64`	the new server's IPv6 range
`api.example.com`	the API's domain
`old-box`, `new-box`	SSH aliases for the two servers (set up in part 2)

What we are moving

Only Caddy publishes ports to the outside world. Everything else talks over
Docker's private network. That detail comes back several times, because it
decides what the firewall has to allow and what the migration has to copy.

1. Picking a server type

The three families

When you click Add Server, the Type step splits the catalogue into
Shared Resources (two families) and Dedicated Resources (one):

Family	Series	CPU	Hetzner's own pitch	Who it's for
Cost-Optimized (shared)	CX (x86) and CAX (Arm64)	shared Intel/AMD, older hardware generations	"cost-efficient… with limited availability"	small production services, side projects. Cheapest per GB of RAM
Regular Performance (shared)	CPX	shared AMD, newer hardware generations	"best price/performance"	the same jobs with faster, more consistent CPU
General Purpose (dedicated)	CCX	dedicated AMD vCPUs, latest hardware	"predictable performance, critical production"	sustained CPU load: builds, busy databases

Here is each family as the console showed it in Helsinki on the day I
created the server (October 2026, USD per month, excluding VAT):

Cost-Optimized. CX23 is 2 vCPU, 4 GB, 40 GB for $6.49. Note that every row is greyed out: at that moment none of them could be ordered in Helsinki. The
summary on the right still says CPX22 because a greyed row can't be selected.

Regular Performance. CPX22 has the same 2 vCPU and 4 GB as CX23 (with 80 GB of disk) for $22.99, more than three times the price.

General Purpose. Dedicated cores start at CCX13 (2 vCPU, 8 GB) for $50.49.

The panel on the right shows one cost that's easy to miss: a public IPv4 is billed on top of the server, $0.60/month.

Cost-Optimized also offers Arm64 (Ampere), which is often the best value of all. I couldn't use it: our container image is built for amd64 only.

Check your image's architecture before choosing Arm. docker manifest inspect <image> lists the platforms it was built for.

"Shared" means your virtual CPU is a time slice of a physical core that other
customers' VMs use too. Most of the time you get the whole slice. When a
neighbour is busy you may get less, which is called CPU steal (shown as st
in top). For an API that idles most of the day that's fine; for a CI runner,
it isn't. Dedicated vCPUs are reserved for you, which is what the large price
jump buys.

How big does it have to be?

Measure before you buy. On the old server, docker stats --no-stream and
free -h showed the entire stack using about 1 GB. On the fresh Hetzner box,
Ubuntu itself sat at about 500 MB used after boot.

Option (Helsinki, October 2026)	vCPU / RAM / disk	Monthly + IPv4	Verdict
CPX02 (Regular Performance)	1 / 1 GB / 20 GB	$7.49 + $0.60	too small: Postgres, Valkey, the server and Caddy together would spill into swap
CPX12 (Regular Performance)	1 / 2 GB / 40 GB	$13.49 + $0.60	tight on RAM, and double CX23's price
CX23 (Cost-Optimized)	2 / 4 GB / 40 GB	$6.49 + $0.60	chosen: 4× headroom on RAM, the cheapest option that fits
CPX22 (Regular Performance)	2 / 4 GB / 80 GB	$22.99 + $0.60	the fallback if CX23 stays unavailable
CCX13 (General Purpose)	2 dedicated / 8 GB / 80 GB	$50.49 + $0.60	far more than an idle API needs

Prices change; take today's from the console, not from this table.

"Limited availability" is real

Cost-Optimized types carry a limited availability label, and it means it. When
I first configured the server, CX23 in Helsinki was available. A few minutes later,
when I came back to click Create, the whole CX row set was greyed out: not available (the first screenshot above)..

Hetzner allocates these types from a pool of hosts per location. When a
location's pool is full, the type disappears there until capacity frees up.
Your options:

Change the location. Nuremberg (nbg1) and Falkenstein (fsn1) in Germany and Helsinki (hel1) in Finland are all in the eu-central network zone, with similar latency to most of Europe. The type is often in stock at one of the others.
Wait and retry. Capacity comes and goes during the day. My CX23 in Helsinki came back later.
Pick the next family up (CPX), if you can't wait and the price is acceptable.

A consequence I didn't expect: once you hold a scarce server type, deleting the
server to fix a mistake is risky. You may not get another one. That shaped
part 2, where I fixed a setup mistake on the running server instead of
recreating it.

Location and image

Location: as close as possible to your users and to your object storage. My storage is a global-anycast S3 provider, so any EU location was fine.
Image: Ubuntu 26.04 LTS. "LTS" means five years of security updates, which is what you want on a server you'd rather not rebuild. 24.04 LTS would have been equally good.

2. SSH keys: how you'll get in

What a key pair is

An SSH key pair is two halves of one mathematical object:

the private key (~/.ssh/hetzner_ed25519) stays on your laptop and never leaves it;
the public key (~/.ssh/hetzner_ed25519.pub) goes on every server you want to log in to, in ~/.ssh/authorized_keys.

The server never learns your private key. At login it asks your SSH client to
sign a value that is unique to this connection. Your client signs it with the
private key, and the server checks the signature against the public key it has
on file. A valid signature proves you hold the private key without the key ever
crossing the network.

Notice there are two key pairs in play:

Key	Belongs to	Proves	Stored where
host key	the server	"I really am 198.51.100.20, not an impostor"	server: `/etc/ssh/ssh_host_*`; your laptop remembers it in `~/.ssh/known_hosts`
user key	you	"I'm allowed to log in as `ubuntu`"	laptop: private key; server: `authorized_keys`

The first time you connect, SSH shows the host key's fingerprint and asks
whether to trust it. This is called TOFU: trust on first use. If the host key
later changes, for example because you rebuilt the server, SSH refuses to connect
and warns about a possible man-in-the-middle. That's correct behaviour. After a
deliberate rebuild, remove the old entry with ssh-keygen -R <host>.

Creating the key

ssh-keygen -t ed25519 -f ~/.ssh/hetzner_ed25519 -C hetzner-api

-t ed25519: a modern elliptic-curve key. It's short (one line), fast, and has no weak-parameter choices to get wrong, unlike older RSA keys.
-f: the file name. A separate key per provider means losing or rotating one doesn't touch the others.
-C: a comment, just a label so you can recognise the key in authorized_keys later.

It asks for a passphrase. Use one: it encrypts the private key on disk, and
ssh-agent remembers it for your session, so you type it once.

Print the public half and paste it into Hetzner → Security → SSH keys → Add
SSH key:

cat ~/.ssh/hetzner_ed25519.pub

When you pick this key on the create screen, Hetzner writes it into the new
server's /root/.ssh/authorized_keys, and, as a side effect, does not set a
root password at all. Remember that; it matters in part 2.

3. Primary IPs: an address that outlives the server

What a Primary IP is

In Hetzner Cloud, a server's public address is a separate resource called a
Primary IP. It's created along with the server and attached to it, but it has
a life of its own:

IPv4: one address, billed separately (about $0.60/month), because the world has run out of IPv4.
IPv6: a whole /64 network, free. That's 2⁶⁴ addresses; the server uses the first one by convention (2001:db8:1234:5678::1).

Why "auto delete" matters

Each Primary IP has an auto delete switch. On, the address is released when
the server is deleted, and you'll probably never get it back. Off, the address
stays in your project, unattached, ready to assign to a replacement server.

Why you want it off for production:

DNS points at it. A new address means a DNS change, and DNS changes take a while to reach everyone (part 3 explains TTLs).
Other services may allowlist it. Object-storage tokens, database firewalls and payment-provider webhook settings sometimes filter by source IP.
It's your escape hatch. If the server breaks beyond repair, you can delete it and create a new one with the same address.

Turn it off in IPs → (the IPv4) → ⋯. If the menu offers "Enable Auto
Delete", it's already off.

I didn't add an IPv6 (AAAA) DNS record. The old server had none, so the
migration changes as little as possible. That can come later.

4. The Cloud Firewall: the perimeter outside your VM

Two places a firewall can sit

A Hetzner Cloud Firewall filters traffic in Hetzner's network, before it
reaches your VM. A host firewall like ufw filters it inside the VM.

The difference matters because of Docker. When a container publishes a port
(ports: ["443:443"]), Docker writes its own NAT rules that send the packet to
the container before it reaches the chain ufw controls. So ufw deny 443
does not block a published Docker port, and plenty of people have exposed a
database to the internet this way while ufw status said everything was closed.

The Cloud Firewall can't be bypassed from inside the VM, so it is the real
perimeter. ufw (set up in part 2) is a second layer for anything that isn't
Docker.

Stateful filtering, in one paragraph

Both firewalls are stateful: they track connections. You write rules for
new incoming connections only. Replies to connections your server opened (to
download packages, call an API, fetch a certificate) are recognised as part of
an existing connection and let back in automatically. That's why the rule set
below has no outbound rules at all: with no outbound rules, Hetzner allows all
outbound traffic, and the server needs outbound access to apt, the container
registry, object storage, the email provider and Let's Encrypt.

The rules

Create it before the server (Firewalls → Create Firewall) so the server is
born protected:

Protocol	Port	Source	Why
TCP	2222	Any IPv4, Any IPv6	SSH, on a non-standard port
TCP	22	Any	SSH, for networks that block outbound 2222
TCP	80	Any	Let's Encrypt's HTTP-01 challenge, and the HTTP→HTTPS redirect
TCP	443	Any	the API
UDP	443	Any	HTTP/3 (QUIC runs over UDP)
ICMP	—	Any	ping, and path-MTU discovery (blocking all ICMP can silently break large packets)

Not on the list: 5432 (Postgres) and 6379 (Valkey). They're never published, and
if a misconfiguration ever did publish them, this firewall would still drop the
traffic.

Why two SSH ports? Port 22 gets a constant stream of automated password-guessing
bots. Moving SSH to 2222 isn't security, since a port scan finds it in seconds,
but it cuts that noise to almost nothing and keeps the logs readable. Key-only
login (part 2) is the actual security. I kept 22 as well, because some networks
block outbound connections to unusual ports. As part 2 shows, my home network
turned out to do the reverse.

5. cloud-init: configuring the server before you first log in

What it is

cloud-init is a program in every major cloud image that runs once, on first
boot. It reads "user data" the provider hands to the VM, which is the text you
paste into the Cloud config box on the create screen, and applies it:
creating users, writing files, installing packages, running commands.

The appeal is that the server is configured before anyone logs in. You never
have a window where it's reachable with default settings, and the configuration
is a document you can review and reuse.

#cloud-config
users:
  - name: ubuntu
    groups: [sudo]
    shell: /bin/bash
    sudo: "ALL=(ALL) NOPASSWD:ALL"
    ssh_authorized_keys:
      - ssh-ed25519 AAAA...your-public-key... hetzner-api
disable_root: true
ssh_pwauth: false
package_update: true
package_upgrade: true
packages: [ufw, unattended-upgrades]
write_files:
  - path: /etc/ssh/sshd_config.d/10-hardening.conf
    content: |
      Port 22
      Port 2222
      PasswordAuthentication no
      KbdInteractiveAuthentication no
      PermitRootLogin no
      PubkeyAuthentication yes
  - path: /etc/docker/daemon.json
    content: |
      {"log-driver": "json-file", "log-opts": {"max-size": "10m", "max-file": "3"}}
runcmd:
  - ufw default deny incoming
  - ufw default allow outgoing
  - ufw allow 22/tcp
  - ufw allow 2222/tcp
  - ufw allow 80/tcp
  - ufw allow 443/tcp
  - ufw allow 443/udp
  - ufw --force enable
  - systemctl daemon-reload
  - systemctl restart ssh.socket || true
  - systemctl restart ssh || true
  - fallocate -l 2G /swapfile && chmod 600 /swapfile && mkswap /swapfile && swapon /swapfile
  - echo '/swapfile none swap sw 0 0' >> /etc/fstab

Every line here is explained in part 2, where we apply the same settings by hand.

The order cloud-init works in

One thing that will surprise you if you SSH in too early: cloud-init runs its
stages in order, and package_upgrade comes before runcmd. On a fresh image
the upgrade can take a few minutes, and during that time port 2222 isn't open yet.
cloud-init status --wait blocks until everything has finished.

What actually happened

I created the server with the type, location, image, SSH key and firewall all
set, and forgot to paste the cloud config. The box came up as a stock image:
only root, only port 22, no ubuntu user, no ufw, no swap. My first
ssh -p 2222 got Connection refused.

"Refused" is useful information. It means the packet reached the machine (the
Cloud Firewall allowed it) and the kernel answered "nothing is listening here".
A firewall drop would have shown up as a timeout instead. Part 2 goes into
that difference.

Since CX23 was close to sold out, I didn't delete and recreate. I applied the same
configuration by hand, as root over port 22. That's part 2.

6. Create

The final create screen, for reference:

Field	Value
Location	Helsinki (or wherever the type is in stock)
Image	Ubuntu 26.04 LTS
Type	Shared vCPU → Cost-Optimized → x86 → CX23
Networking	Public IPv4 ✅, Public IPv6 ✅
SSH keys	`hetzner-api`
Firewalls	the firewall from section 4
Backups	off (the database is backed up to object storage; see part 3)
Cloud config	the YAML above. Don't forget it
Name	something meaningful. The default `ubuntu-4gb-hel1-1` tells you nothing in six months

Then turn off auto delete on the new IPv4 (section 3), and check that the
firewall shows "Fully applied" with one server.

What we have so far

a CX23 with 4 GB RAM for about $7/month including IPv4;
an IPv4 address that survives the server;
a perimeter firewall that allows only SSH, HTTP(S) and ICMP;
a stock Ubuntu, reachable as root on port 22 only, because the cloud config was never applied.

In part 2 we fix that by hand: a non-root user, key-only SSH on two ports,
ufw, swap, automatic security updates and Docker. Along the way: how to configure
two servers in ~/.ssh/config, how to make changes over SSH without locking
yourself out, and how to tell from a port test whether a firewall or a closed
port is in your way.

One .env file, three answers: measuring how dotenv readers disagree

Ajit Kumar — Tue, 27 Oct 2026 13:40:20 +0000

A .env file looks like the simplest config format there is. It has no spec, though. Every loader implements its own dialect, and the dialects disagree on a handful of edge cases. The same line can give different values to a Docker container and a Python app in the same project, and nothing tells you.

This post shows the problem with code. We start with one innocent line, build a small conformance harness that feeds a corpus of .env fixtures to six readers, and look at what comes out. Then we use differential fuzzing to find disagreements nobody listed.

All code, fixtures and result files are in the companion repo: urwithajit9/dotenv-difference.

What we found, in short:

13 of 16 fixtures produce different values depending on the reader. Only bare keys, duplicate keys and an empty file come out identical everywhere.
One line gets three different values. API_KEY= # fill this in later is read as the comment text, as the comment text with leading spaces, or as an empty string.
Compose and docker run --env-file are different parsers. They disagree on 9 of 16 fixtures.
Node silently truncates at any #. Tr0ub4dor#3 becomes Tr0ub4dor.
Fuzzing finds more. Nearly half of random one-line files are read differently by python-dotenv and Node, and some of the causes (name grammar, export =v) are not in any list of known differences.

Measured environment

Everything marked measured was run for this post, on one machine, against the same 16 fixtures.

Component	Version
Python	3.14.4
python-dotenv	1.2.2
Node	v22.22.1
`dotenv` (npm)	18.0.9
`dotenv-expand` (npm)	12.0.1
bash	GNU bash 5.x (`bash --version`)
Docker	29.6.2 (build dfc4efb)
Docker Compose	v5.3.1

Not measured: godotenv, dotenvy and Ruby dotenv. The harness is built so you can add them (see "Extending the harness"). The evnx 0.10.18 value in the first table comes from an earlier run of the same line on 2026-10-27 and is not part of the harness.

1. The line that started this

API_KEY=   # fill this in later

The author's intent is clear: the key is empty, and the comment is a note to self. What does each reader think?

Reader	Value of `API_KEY`	Status
Docker Compose v5.3.1	`# fill this in later`	measured
python-dotenv 1.2.2	`# fill this in later`	measured
evnx 0.10.18	`# fill this in later`	measured (earlier run)
`docker run --env-file`	`# fill this in later` (three leading spaces kept)	measured
Node dotenv 18.0.9	`""` (empty)	measured
bash `source`	`""` (empty)	measured

Three different values. Compose and python-dotenv keep the comment text. docker run --env-file keeps the comment text and the whitespace in front of it. Node and bash treat the value as empty. Anything that reads this file and checks "is API_KEY set?" gets a different answer depending on the reader.

A correction worth making: an earlier draft of the underlying issue assumed python-dotenv returns an empty string here. It doesn't. It agrees with Compose. That is exactly why you should measure instead of writing the table from memory.

2. A tiny harness

The idea: one directory of fixtures, one adapter per reader that reads a fixture (or a whole directory) and prints what it sees as JSON, and a script that builds a comparison table from the JSON files.

dotenv-difference/
├── make_cases.py          # writes cases/*.env
├── readers/
│   ├── py_dotenv.py       # python-dotenv, one file
│   ├── py_dotenv_all.py   # python-dotenv, whole directory
│   ├── node_dotenv_all.js # Node dotenv (+ dotenv-expand)
│   ├── bash_source.sh     # bash `source`
│   └── docker_env.sh      # Compose and `docker run --env-file`
├── compare.py             # results/*.json -> comparison table
├── fuzz.py                # differential fuzzing
├── run_all.sh             # regenerate everything
└── results/               # JSON output, committed

Fixtures

Each fixture is a small file named for the case. Create them with a script so the bytes are exact (CRLF and BOM are easy to mangle in an editor, and printf behaves differently across shells):

# make_cases.py
from pathlib import Path

CASES = {
    "empty_comment":  b"API_KEY=   # fill this in later\n",
    "empty_comment2": b"EMPTY_C2= #c\n",
    "value_comment":  b"V_C=x # c\n",
    "hash_in_value":  b"HASH=Tr0ub4dor#3\n",
    "hash_start":     b"COLOR=#ff0000\n",
    "bare":           b"BARE=\n",
    "export":         b"export KEY=v\n",
    "expand":         b"HOST=db\nURL=${HOST}:5432\n",
    "escapes":        b'A="a\\nb"\nB=\'a\\nb\'\n',
    "dup":            b"K=1\nK=2\n",
    "spaces":         b"KEY = v \nK2=v   \n",
    "multiline":      b'PEM="-----BEGIN\nabc\n-----END"\n',
    "crlf":           b"K=v\r\nL=w\r\n",
    "bom":            b"\xef\xbb\xbfK=v\n",
    "empty_file":     b"",
    "unterminated":   b'K="unterminated\nL=next\n',
}

out = Path("cases")
out.mkdir(exist_ok=True)
for name, data in CASES.items():
    (out / f"{name}.env").write_bytes(data)

Run it. It prints nothing, but it creates a cases/ directory with 16 fixtures:

user@host:~/dotenv-difference$ python3 make_cases.py
user@host:~/dotenv-difference$ ls cases/
bare.env           bom.env             crlf.env          dup.env
empty_comment.env  empty_comment2.env  empty_file.env    escapes.env
expand.env         export.env          hash_in_value.env hash_start.env
multiline.env      spaces.env          unterminated.env  value_comment.env

3. Readers and adapters

Each adapter takes a file path or a directory and prints a JSON object of what that reader sees.

Python: python-dotenv

Create a virtual environment (recent Linux distributions block system-wide pip install) and pin the version used in this post:

user@host:~/dotenv-difference$ python3 -m venv .venv
user@host:~/dotenv-difference$ source .venv/bin/activate
(.venv) user@host:~/dotenv-difference$ pip install python-dotenv==1.2.2

The smallest possible adapter:

# readers/py_dotenv.py
import sys, json
from dotenv import dotenv_values
print(json.dumps(dict(dotenv_values(sys.argv[1])), ensure_ascii=False))

It needs the fixture path as an argument. Run it without one and you get an IndexError:

(.venv) user@host:~/dotenv-difference$ python3 readers/py_dotenv.py cases/empty_comment.env
{"API_KEY": "# fill this in later"}

python-dotenv keeps the comment as the value, which is the first surprise in this post.

For the whole fixture directory, use this version. It writes one JSON file and records a failure instead of crashing:

# readers/py_dotenv_all.py
import argparse, json, sys
from pathlib import Path
from dotenv import dotenv_values


def read(path):
    try:
        return dict(dotenv_values(path))
    except Exception as e:
        # Record the failure instead of crashing the whole run
        return {"error": f"{type(e).__name__}: {e}"}


def main():
    ap = argparse.ArgumentParser(description="Parse .env files with python-dotenv and emit JSON")
    ap.add_argument("path", help="a .env file, or a directory of .env files")
    ap.add_argument("--out", help="write JSON to this file instead of stdout")
    args = ap.parse_args()

    p = Path(args.path)
    if p.is_dir():
        result = {f.stem: read(f) for f in sorted(p.glob("*.env"))}
    else:
        result = read(p)

    # ensure_ascii stays on, so a BOM shows up as \ufeff instead of an invisible character
    text = json.dumps(result, indent=2, sort_keys=True) + "\n"
    if args.out:
        Path(args.out).parent.mkdir(parents=True, exist_ok=True)
        Path(args.out).write_text(text, encoding="utf-8")
    else:
        sys.stdout.write(text)


if __name__ == "__main__":
    main()

(.venv) user@host:~/dotenv-difference$ python3 readers/py_dotenv_all.py cases/ --out results/python-dotenv.json
python-dotenv could not parse statement starting at line 1

The warning comes from the unterminated fixture (K="unterminated). It is the only sign that a line was rejected, and it goes to stderr, so it is easy to miss in a script or CI log.

results/python-dotenv.json (shown compactly; the file in the repo is indented):

{
  "bare": { "BARE": "" },
  "bom": { "\ufeffK": "v" },
  "crlf": { "K": "v", "L": "w" },
  "dup": { "K": "2" },
  "empty_comment": { "API_KEY": "# fill this in later" },
  "empty_comment2": { "EMPTY_C2": "#c" },
  "empty_file": {},
  "escapes": { "A": "a\nb", "B": "a\\nb" },
  "expand": { "HOST": "db", "URL": "db:5432" },
  "export": { "KEY": "v" },
  "hash_in_value": { "HASH": "Tr0ub4dor#3" },
  "hash_start": { "COLOR": "#ff0000" },
  "multiline": { "PEM": "-----BEGIN\nabc\n-----END" },
  "spaces": { "K2": "v", "KEY": "v" },
  "unterminated": { "L": "next" },
  "value_comment": { "V_C": "x" }
}

What to notice:

empty_comment: the value is the comment text, not an empty string.
bom: the key is \ufeffK, not K. A lookup for K fails, even though the file looks fine in an editor.
unterminated: K is gone, with only a log line. Only L survives.
expand: ${HOST} was resolved to db by default.
escapes: \n becomes a real newline in double quotes and stays literal in single quotes.
dup: the last value wins.

Node: dotenv, with and without dotenv-expand

Expansion is a separate package in the Node world, so we report both. Check your Node version, create a package.json, and install both packages at exact versions:

user@host:~/dotenv-difference$ node --version
v22.22.1
user@host:~/dotenv-difference$ npm init -y
user@host:~/dotenv-difference$ npm install --save-exact dotenv@18.0.9 dotenv-expand@12.0.1
added 3 packages, and audited 4 packages in 424ms
user@host:~/dotenv-difference$ echo "node_modules/" >> .gitignore

Pinning matters. Results depend on the parser version, and --save-exact stops a later npm install from silently moving to a newer release.

// readers/node_dotenv_all.js
// Usage: node readers/node_dotenv_all.js <file-or-dir> [--out results/node-dotenv.json]
const fs = require("fs");
const path = require("path");
const dotenv = require("dotenv");
const { expand } = require("dotenv-expand");

function read(file) {
  try {
    const plain = dotenv.parse(fs.readFileSync(file)); // pass the raw Buffer
    // processEnv: {} so expansion can't see the real environment
    const expanded = expand({ parsed: { ...plain }, processEnv: {} }).parsed;
    return { plain, expanded };
  } catch (e) {
    return { error: `${e.name}: ${e.message}` };
  }
}

// Stable key order, so the file diffs cleanly in git
function sortKeys(value) {
  if (Array.isArray(value)) return value.map(sortKeys);
  if (value && typeof value === "object") {
    return Object.fromEntries(
      Object.keys(value).sort().map((k) => [k, sortKeys(value[k])])
    );
  }
  return value;
}

const args = process.argv.slice(2);
const outIdx = args.indexOf("--out");
const out = outIdx !== -1 ? args[outIdx + 1] : null;
const target = args.find((a, i) => !a.startsWith("--") && i !== outIdx + 1);

if (!target) {
  console.error("usage: node readers/node_dotenv_all.js <file-or-dir> [--out file.json]");
  process.exit(2);
}

let result;
if (fs.statSync(target).isDirectory()) {
  result = {};
  for (const f of fs.readdirSync(target).filter((f) => f.endsWith(".env")).sort()) {
    result[path.basename(f, ".env")] = read(path.join(target, f));
  }
} else {
  result = read(target);
}

const text = JSON.stringify(sortKeys(result), null, 2) + "\n";
if (out) {
  fs.mkdirSync(path.dirname(out), { recursive: true });
  fs.writeFileSync(out, text, "utf8");
} else {
  process.stdout.write(text);
}

user@host:~/dotenv-difference$ node readers/node_dotenv_all.js cases/empty_comment.env
{ "expanded": { "API_KEY": "" }, "plain": { "API_KEY": "" } }
user@host:~/dotenv-difference$ node readers/node_dotenv_all.js cases/ --out results/node-dotenv.json

Nothing is printed for the whole-directory run, and no warnings appear, even for the malformed unterminated fixture. Node's parser accepts it silently.

The output has two objects per case. plain is what dotenv alone returns. expanded is the result after dotenv-expand resolves ${VAR} references. They differ only on the expand fixture:

"expand": {
  "expanded": { "HOST": "db", "URL": "db:5432" },
  "plain":    { "HOST": "db", "URL": "${HOST}:5432" }
}

Bash: `source`

Bash is different: it executes the file instead of parsing it. We run it in an empty environment (env -i), compare the environment before and after, and report what changed, plus the exit code and any error message.

Warning: source executes whatever is in the file. These fixtures are harmless, but a real .env can contain $(...) or backticks. Never run this on a file you don't trust.

#!/usr/bin/env bash
# readers/bash_source.sh
# Usage: readers/bash_source.sh <file-or-dir>   (prints JSON on stdout)
# WARNING: this EXECUTES each file with bash. Only run it on fixtures you trust.
set -u

# Escape a string for use inside a JSON "..." value
json_escape() {
  local s=$1
  s=${s//\\/\\\\}
  s=${s//\"/\\\"}
  s=${s//$'\n'/\\n}
  s=${s//$'\r'/\\r}
  s=${s//$'\t'/\\t}
  printf '%s' "$s"
}

# Load NUL-separated KEY=VALUE records from stdin into the array named $1
load_env() {
  local -n _arr=$1
  local kv
  while IFS= read -r -d '' kv; do
    [[ $kv == *=* ]] && _arr[${kv%%=*}]=${kv#*=}
  done
}

# Variables bash sets by itself, plus our helper variables
declare -A IGNORE=([SHLVL]=1 [_]=1 [PWD]=1 [OLDPWD]=1 [rc]=1 [__RC__]=1)

# Baseline: what an empty-environment bash already has
declare -A BASE=()
load_env BASE < <(env -i bash -c 'env -0')

read_one() {
  local file=$1 errfile err key rc=0 first=1
  declare -A after=()
  errfile=$(mktemp)

  # Run in an empty environment; set -a exports everything the file assigns.
  load_env after < <(
    env -i bash -c 'set -a; source "$1"; rc=$?; set +a; env -0; printf "__RC__=%s\0" "$rc"' _ "$file" 2>"$errfile"
  )
  rc=${after[__RC__]:-?}
  err=$(<"$errfile"); rm -f "$errfile"

  printf '{\n  "exit": %s,\n' "$rc"
  [[ -n $err ]] && printf '  "stderr": "%s",\n' "$(json_escape "$err")"
  printf '  "values": {'
  while IFS= read -r key; do
    [[ -z $key || -n ${IGNORE[$key]:-} ]] && continue
    [[ ${BASE[$key]-__unset__} == "${after[$key]}" ]] && continue
    (( first )) || printf ','
    first=0
    printf '\n    "%s": "%s"' "$(json_escape "$key")" "$(json_escape "${after[$key]}")"
  done < <(printf '%s\n' "${!after[@]}" | sort)
  (( first )) && printf '}\n}' || printf '\n  }\n}'
}

target=${1:-}
[[ -z $target ]] && { echo "usage: $0 <file-or-dir>" >&2; exit 2; }
target=${target%/}   # avoid "cases//x.env" in error messages

if [[ -d $target ]]; then
  printf '{'
  sep=''
  for f in "$target"/*.env; do
    name=$(basename "$f" .env)
    printf '%s\n  "%s": ' "$sep" "$name"
    read_one "$f" | sed '2,$s/^/  /'
    sep=','
  done
  printf '\n}\n'
else
  read_one "$target"; printf '\n'
fi

Bash 4.3 or newer is required (the script uses namerefs). Make it executable and run it:

user@host:~/dotenv-difference$ chmod +x readers/bash_source.sh
user@host:~/dotenv-difference$ ./readers/bash_source.sh cases/spaces.env
{
  "exit": 0,
  "stderr": "cases/spaces.env: line 1: KEY: command not found",
  "values": {
    "K2": "v"
  }
}
user@host:~/dotenv-difference$ ./readers/bash_source.sh cases/ > results/bash-source.json

In spaces.env, KEY = v is run as a command named KEY, so it fails. The next line, K2=v, still loads.

Docker: Compose and `docker run --env-file`

Docker has two separate parsers for .env files, and they don't behave the same:

Docker Compose reads the file through env_file: in a Compose file.
docker run --env-file uses a different, simpler parser.

The adapter loads each fixture through one of them, starts a throwaway container, and prints the environment the container received. That is the value your app would see. It runs env -0, so multi-line values survive, and subtracts the variables the image sets itself (PATH, HOME, and so on).

#!/usr/bin/env bash
# readers/docker_env.sh
# Usage: readers/docker_env.sh <compose|run> <file-or-dir>   (prints JSON on stdout)
#   compose : Docker Compose reads the file via env_file, then we print the container's env
#   run     : `docker run --env-file` reads the file, then we print the container's env
# Needs: docker (and the compose plugin for "compose"). The image is pulled on first use.
set -u
IMAGE=${IMAGE:-alpine:3.20}

mode=${1:-}; target=${2:-}
if [[ $mode != compose && $mode != run || -z $target ]]; then
  echo "usage: $0 <compose|run> <file-or-dir>" >&2; exit 2
fi
command -v docker >/dev/null || { echo "docker not found" >&2; exit 2; }
target=${target%/}

json_escape() {
  local s=$1
  s=${s//\\/\\\\}; s=${s//\"/\\\"}
  s=${s//$'\n'/\\n}; s=${s//$'\r'/\\r}; s=${s//$'\t'/\\t}
  printf '%s' "$s"
}

load_env() {
  local -n _arr=$1
  local kv
  while IFS= read -r -d '' kv; do
    [[ $kv == *=* ]] && _arr[${kv%%=*}]=${kv#*=}
  done
}

# Variables the image/container sets by itself (HOSTNAME changes on every run)
declare -A IGNORE=([HOSTNAME]=1 [__RC__]=1)
declare -A BASE=()
load_env BASE < <(docker run --rm "$IMAGE" env -0)

# Run one fixture through the chosen reader; prints "env -0" of the container.
reader_cmd() {
  local file=$1
  if [[ $mode == run ]]; then
    docker run --rm --env-file "$file" "$IMAGE" env -0
  else
    local abs proj; abs=$(realpath "$file"); proj=$(mktemp -d)
    cat > "$proj/compose.yml" <<YAML
services:
  t:
    image: $IMAGE
    network_mode: none
    env_file: ["$abs"]
YAML
    docker compose -p dotenv_diff -f "$proj/compose.yml" run --rm --no-deps -T t env -0
    local rc=$?
    rm -rf "$proj"
    return $rc
  fi
}

read_one() {
  local file=$1 errfile err key rc first=1 abs
  declare -A after=()
  abs=$(realpath "$file"); errfile=$(mktemp)

  load_env after < <( { reader_cmd "$file" 2>"$errfile"; printf '__RC__=%s\0' "$?"; } )
  rc=${after[__RC__]:-?}
  err=$(<"$errfile"); rm -f "$errfile"
  err=${err//"$abs"/$file}          # keep absolute paths (and usernames) out of the output
  err=${err//"$HOME"/\~}
  # Drop Compose's progress lines: they contain a random container name, which would
  # make the results file change on every run
  err=$(grep -v -E '^ ?(Container|Network|Volume) .* (Creating|Created|Starting|Started|Removing|Removed|Running|Waiting|Healthy) ?$' <<<"$err")

  printf '{\n  "exit": %s,\n' "$rc"
  [[ -n $err ]] && printf '  "stderr": "%s",\n' "$(json_escape "$err")"
  printf '  "values": {'
  while IFS= read -r key; do
    [[ -z $key || -n ${IGNORE[$key]:-} ]] && continue
    [[ ${BASE[$key]-__unset__} == "${after[$key]}" ]] && continue
    (( first )) || printf ','
    first=0
    printf '\n    "%s": "%s"' "$(json_escape "$key")" "$(json_escape "${after[$key]}")"
  done < <(printf '%s\n' "${!after[@]}" | sort)
  (( first )) && printf '}\n}' || printf '\n  }\n}'
}

if [[ -d $target ]]; then
  printf '{'
  sep=''
  for f in "$target"/*.env; do
    printf '%s\n  "%s": ' "$sep" "$(basename "$f" .env)"
    read_one "$f" | sed '2,$s/^/  /'
    sep=','
  done
  printf '\n}\n'
else
  read_one "$target"; printf '\n'
fi

Check your versions, make the script executable, and run the line the whole article is about through both parsers:

user@host:~/dotenv-difference$ docker --version
Docker version 29.6.2, build dfc4efb
user@host:~/dotenv-difference$ docker compose version
Docker Compose version v5.3.1
user@host:~/dotenv-difference$ chmod +x readers/docker_env.sh

user@host:~/dotenv-difference$ ./readers/docker_env.sh compose cases/empty_comment.env
{
  "exit": 0,
  "values": {
    "API_KEY": "# fill this in later"
  }
}

user@host:~/dotenv-difference$ ./readers/docker_env.sh run cases/empty_comment.env
{
  "exit": 0,
  "values": {
    "API_KEY": "   # fill this in later"
  }
}

Same file, same line, two different values. Compose keeps # fill this in later. docker run --env-file keeps the three leading spaces too. (The first run pulls the alpine:3.20 image, so it takes a few seconds longer.)

To see what Compose resolved without starting a container:

user@host:~/dotenv-difference$ docker compose -f - config --format json <<EOF | jq '.services.t.environment'
services:
  t:
    image: alpine:3.20
    env_file: ["$PWD/cases/empty_comment.env"]
EOF
{
  "API_KEY": "# fill this in later"
}

config and the container agree. Now every fixture, through both parsers:

user@host:~/dotenv-difference$ ./readers/docker_env.sh compose cases/ > results/compose.json
user@host:~/dotenv-difference$ ./readers/docker_env.sh run cases/ > results/docker-run.json

Each fixture starts one container, so expect it to take a minute or so.

4. Comparing the readers automatically

Six JSON files are hard to compare by eye, so a small script loads them and builds the table. Because it reads the result files, the table can't drift from the data. It skips any reader whose results file doesn't exist yet, so it also works with just the first few adapters.

#!/usr/bin/env python3
"""Build the reader-comparison table from results/*.json.

Usage:
  python3 compare.py                 # only the cases where readers differ
  python3 compare.py --all           # every case
  python3 compare.py --cases cases/  # label rows with the fixture's content
"""
import argparse, json, sys
from pathlib import Path

# column name -> (results file, how to pull the values out of one case entry,
#                 whether the entry has an "exit" status to flag)
COLUMNS = {
    "python-dotenv":           ("python-dotenv.json", lambda e: e,              False),
    "Node dotenv":             ("node-dotenv.json",   lambda e: e["plain"],     False),
    "Node + expand":           ("node-dotenv.json",   lambda e: e["expanded"],  False),
    "bash `source`":           ("bash-source.json",   lambda e: e["values"],    True),
    "Compose":                 ("compose.json",       lambda e: e["values"],    True),
    "`docker run --env-file`": ("docker-run.json",    lambda e: e["values"],    True),
}


def load(results_dir):
    files = {}
    for fname, _, _ in COLUMNS.values():
        if fname not in files:
            files[fname] = json.loads((results_dir / fname).read_text(encoding="utf-8"))
    return files


def values_of(files, col, case):
    fname, pick, _ = COLUMNS[col]
    return pick(files[fname][case])


def failed(files, col, case):
    fname, _, has_exit = COLUMNS[col]
    return has_exit and files[fname][case]["exit"] != 0


def vis(s):
    # Make invisible characters visible (CR, LF, BOM) and escape quotes
    return json.dumps(s)[1:-1]


def cell(values, is_failed):
    if values:
        text = ", ".join(f'{vis(k)}="{vis(v)}"' for k, v in values.items())
    else:
        text = "(nothing set)"
    text = text.replace("|", "\\|")
    if is_failed:
        text += " ⚠ error"
    return f"`{text}`"


def label(cases_dir, case):
    if not cases_dir:
        return f"`{case}`"
    raw = (cases_dir / f"{case}.env").read_bytes().decode("utf-8", errors="replace")
    raw = raw.replace("\ufeff", "<BOM>").replace("\r", "<CR>").rstrip("\n")
    raw = raw.replace("\n", " ⏎ ") or "(empty file)"
    return f"`{raw}`".replace("|", "\\|")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--results", default="results")
    ap.add_argument("--cases")
    ap.add_argument("--all", action="store_true")
    a = ap.parse_args()

    # Skip readers whose results file doesn't exist yet
    for name in [n for n, (f, _, _) in COLUMNS.items() if not (Path(a.results) / f).exists()]:
        print(f"note: skipping {name} (no results file yet)", file=sys.stderr)
        del COLUMNS[name]

    files = load(Path(a.results))
    cases = sorted(files["python-dotenv.json"])
    cases_dir = Path(a.cases) if a.cases else None

    rows, agree = [], []
    for case in cases:
        vals = {c: values_of(files, c, case) for c in COLUMNS}
        fails = {c: failed(files, c, case) for c in COLUMNS}
        signature = {json.dumps(v, sort_keys=True) for v in vals.values()}
        differs = len(signature) > 1 or any(fails.values())
        row = [label(cases_dir, case)] + [cell(vals[c], fails[c]) for c in COLUMNS]
        (rows if (differs or a.all) else agree).append((case, row, differs))

    print("| Fixture | " + " | ".join(COLUMNS) + " |")
    print("|---|" + "---|" * len(COLUMNS))
    for _, row, _ in rows:
        print("| " + " | ".join(row) + " |")

    n_diff = sum(1 for _, _, d in rows if d)
    print(f"\n{n_diff} of {len(cases)} fixtures give different results across readers.")
    if agree:
        print("Identical everywhere: " + ", ".join(f"`{c}`" for c, _, _ in agree) + ".")


if __name__ == "__main__":
    main()

user@host:~/dotenv-difference$ python3 compare.py --cases cases/ > results/comparison.md

A fixture counts as "different" if any two columns disagree, or if a reader exits with an error. Pass --all to include the fixtures where every reader agrees.

Regenerating everything

Every step above is a single command, so one script reruns the whole pipeline:

#!/usr/bin/env bash
# run_all.sh - regenerate every fixture, result file and the comparison table
# Needs: the Python venv active, `npm install` done, Docker running.
set -euo pipefail
cd "$(dirname "$0")"

python3 make_cases.py
mkdir -p results

python3 readers/py_dotenv_all.py cases/ --out results/python-dotenv.json
node readers/node_dotenv_all.js cases/ --out results/node-dotenv.json
./readers/bash_source.sh cases/ > results/bash-source.json
./readers/docker_env.sh compose cases/ > results/compose.json
./readers/docker_env.sh run cases/ > results/docker-run.json

python3 compare.py --cases cases/ > results/comparison.md

Commit results/*.json and results/comparison.md. When you bump a dependency, rerun and look at the diff. A changed cell means a loader changed behavior, and you find out in code review instead of in production.

Extending the harness

Every reader is a small adapter that prints {"exit": ..., "values": {...}} as JSON, and compare.py adds a column for each new results file. godotenv (godotenv.Read), dotenvy (dotenvy::from_path_iter) and Ruby dotenv (Dotenv.parse) follow the same pattern, but they were not measured for this post. Pull requests with those adapters are welcome.

5. Results

Six readers, seven columns (Node appears twice, with and without dotenv-expand), sixteen fixtures:

Fixture	python-dotenv	Node dotenv	Node + expand	bash `source`	Compose	`docker run --env-file`
`<BOM>K=v`	`\ufeffK="v"`	`K="v"`	`K="v"`	`(nothing set) ⚠ error`	`K="v"`	`K="v"`
`K=v<CR> ⏎ L=w<CR>`	`K="v", L="w"`	`K="v", L="w"`	`K="v", L="w"`	`K="v\r", L="w\r"`	`K="v", L="w"`	`K="v", L="w"`
`API_KEY= # fill this in later`	`API_KEY="# fill this in later"`	`API_KEY=""`	`API_KEY=""`	`API_KEY=""`	`API_KEY="# fill this in later"`	`API_KEY=" # fill this in later"`
`EMPTY_C2= #c`	`EMPTY_C2="#c"`	`EMPTY_C2=""`	`EMPTY_C2=""`	`EMPTY_C2=""`	`EMPTY_C2="#c"`	`EMPTY_C2=" #c"`
`A="a\nb" ⏎ B='a\nb'`	`A="a\nb", B="a\\nb"`	`A="a\nb", B="a\\nb"`	`A="a\nb", B="a\\nb"`	`A="a\\nb", B="a\\nb"`	`A="a\nb", B="a\\nb"`	`A="\"a\\nb\"", B="'a\\nb'"`
`HOST=db ⏎ URL=${HOST}:5432`	`HOST="db", URL="db:5432"`	`HOST="db", URL="${HOST}:5432"`	`HOST="db", URL="db:5432"`	`HOST="db", URL="db:5432"`	`HOST="db", URL="db:5432"`	`HOST="db", URL="${HOST}:5432"`
`export KEY=v`	`KEY="v"`	`KEY="v"`	`KEY="v"`	`KEY="v"`	`KEY="v"`	`(nothing set) ⚠ error`
`HASH=Tr0ub4dor#3`	`HASH="Tr0ub4dor#3"`	`HASH="Tr0ub4dor"`	`HASH="Tr0ub4dor"`	`HASH="Tr0ub4dor#3"`	`HASH="Tr0ub4dor#3"`	`HASH="Tr0ub4dor#3"`
`COLOR=#ff0000`	`COLOR="#ff0000"`	`COLOR=""`	`COLOR=""`	`COLOR="#ff0000"`	`COLOR="#ff0000"`	`COLOR="#ff0000"`
`PEM="-----BEGIN ⏎ abc ⏎ -----END"`	`PEM="-----BEGIN\nabc\n-----END"`	`PEM="-----BEGIN\nabc\n-----END"`	`PEM="-----BEGIN\nabc\n-----END"`	`PEM="-----BEGIN\nabc\n-----END"`	`PEM="-----BEGIN\nabc\n-----END"`	`PEM="\"-----BEGIN"`
`KEY = v ⏎ K2=v`	`K2="v", KEY="v"`	`K2="v", KEY="v"`	`K2="v", KEY="v"`	`K2="v"`	`K2="v", KEY="v"`	`(nothing set) ⚠ error`
`K="unterminated ⏎ L=next`	`L="next"`	`K="\"unterminated", L="next"`	`K="\"unterminated", L="next"`	`(nothing set) ⚠ error`	`(nothing set) ⚠ error`	`K="\"unterminated", L="next"`
`V_C=x # c`	`V_C="x"`	`V_C="x"`	`V_C="x"`	`V_C="x"`	`V_C="x"`	`V_C="x # c"`

13 of 16 fixtures give different results. Identical everywhere: bare, dup, empty_file.

How to read the cells: \n is a real newline, \\n is a literal backslash followed by n, \r is a carriage return, and \ufeff is the byte order mark.

6. Observations

1. No two readers agree on everything. Counting differing fixtures out of 16: python-dotenv vs Node, 7; python-dotenv vs bash, 7; Compose vs docker run --env-file, 9. Compose vs python-dotenv is the closest pair, at 2. Picking a different pair doesn't make the problem go away.

2. The empty-value-plus-comment line splits the readers three ways. For API_KEY= # fill this in later:

Value of `API_KEY`	Readers
`# fill this in later`	python-dotenv, Compose
`# fill this in later` (leading spaces kept)	`docker run --env-file`
`""` (empty)	Node dotenv, bash

None of these is the value the author meant. For Docker users the failure is worse, because both Docker parsers turn the comment into the secret.

3. Node cuts at any #, silently. The password Tr0ub4dor#3 becomes Tr0ub4dor, and the colour #ff0000 becomes an empty string. No warning, no error. Every other reader keeps both intact.

4. Compose behaves like python-dotenv. The two agree on 14 of 16 fixtures, including the comment, escape, expansion and export cases. They differ only on the BOM (Compose strips it, python-dotenv keeps it in the key name) and on the unterminated quote (Compose fails, python-dotenv drops the line). For any tool that has to pick a default dialect, that is a useful result: matching Compose also means matching most of what Python apps see.

5. docker run --env-file is a line splitter, not a dotenv parser. It takes everything after the first = literally:

quotes are kept: A="a\nb" becomes the value "a\nb" including the quotation marks;
inline comments are not recognised: V_C=x # c keeps x # c;
there is no variable expansion: ${HOST}:5432 stays as text;
whitespace after = is preserved, which is where the leading spaces in observation 2 come from.

6. Multi-line values break silently in docker run --env-file. The PEM fixture becomes "-----BEGIN, the first line only. The continuation lines contain no =, so Docker appears to treat them as names of host variables to pass through, and since none are set, they vanish with no error. A private key would be truncated without any message.

7. docker run --env-file rejects whole files for common syntax. export KEY=v and KEY = v each make it exit with an error and load nothing, so even the valid K2=v line is lost. This matters because export KEY=v is a normal way to write a file that is also meant to be sourced by a shell. The file that works for your shell and for Compose fails for docker run.

8. Expansion depends on which package you installed. ${HOST} is resolved by python-dotenv, bash and Compose by default, but plain Node dotenv and docker run --env-file leave it as text. Adding dotenv-expand changes the Node answer, so "Node" alone isn't one result.

9. Some differences are invisible. CRLF gives bash values a trailing \r that prints like nothing. A BOM makes python-dotenv's first key \ufeffK, so a lookup for K fails even though the file looks right in an editor. Both Docker parsers and Node handle both cleanly.

10. The readers fail in different ways on a malformed line. For the unterminated quote:

Reader	What happens
python-dotenv	drops `K`, keeps `L`, logs a warning to stderr
Node dotenv	keeps `K` with the stray quote in its value, keeps `L`, no warning
`docker run --env-file`	same as Node: keeps `K` with the stray quote, keeps `L`, no warning
Compose	rejects the whole file, sets nothing
bash	rejects the whole file (exit 2), sets nothing

Only python-dotenv, Compose and bash give any signal that something went wrong, and python's is a log line that is easy to miss.

11. Bash is not a dotenv parser, so its disagreements are real traps. It runs the file as a script, so its results come from shell grammar, not dotenv rules:

API_KEY= # fill this in later is an empty assignment followed by a shell comment, so the value is empty. Node reaches the same value by cutting at any #. The shell only treats # as a comment at the start of a word, which is why it keeps Tr0ub4dor#3 and #ff0000.
Spaces around = turn the line into a command. KEY = v runs a program called KEY, which fails, and the file keeps going.
A BOM becomes part of the command name, so K is never set (exit 127).
An unterminated quote loses the whole file, including the valid L=next line.
Double-quoted \n stays literal, and ${HOST} resolves with no extra package, as does anything else the shell understands, such as $(...).

If anyone on a team loads config with source .env, they are using another dialect, and source also executes any command substitution in the file.

12. The agreement is narrow. Bare keys, duplicate keys (last wins) and an empty file come out the same everywhere. Even a trailing comment after a normal value (V_C=x # c) and a quoted multi-line value split the Docker parser from the rest.

What this means

You can't pick a winner, because there's no specification to check against. The practical rules from the data:

put comments on their own line, not after an empty value;
quote any value that contains #, $ or spaces, but remember that docker run --env-file will keep the quotes;
avoid export, spaces around =, ${VAR} references and multi-line values in any file that is used with docker run --env-file;
keep files UTF-8 without a BOM and with LF line endings;
never assume that two tools reading the same .env see the same values, and treat "it works with docker compose" as no evidence that it works with docker run.

7. Why this matters in practice

The disagreement becomes a problem when a tool acts on a value another tool will read differently:

A validator that counts API_KEY= # note as set won't report the missing secret.
A converter or uploader (to JSON, Kubernetes, a cloud secret store) can ship # fill this in later as the literal secret.
A template engine can paste the comment into generated config.
The app itself may read an empty value while your tooling read a string, so the tool and the app disagree and the author never finds out.
A file that works with docker compose can fail or be truncated with docker run --env-file, so two ways of starting the same container give different configuration.

Which reader is "right" is a poor question, because there's no spec. The useful question is: does any plausible reader disagree with this line? That can be tested, and the harness above does exactly that.

8. Bonus: differential fuzzing

A hand-written corpus only finds the cases you thought of. A fuzzer finds the rest. Generate short random lines from the characters that matter, run both readers on each, and report any line where the results differ.

The fuzzer below compares python-dotenv and Node dotenv. It writes nothing to disk (python-dotenv reads from an in-memory stream, and all lines go to Node in one batch), silences python-dotenv's log output, and sorts the results into classes so the interesting ones stand out.

#!/usr/bin/env python3
"""Differential fuzzing: random one-line .env files through python-dotenv and Node dotenv.

Usage: python3 fuzz.py [seed] [count]      (defaults: seed 0, 2000 lines)
"""
import io, json, logging, random, subprocess, sys
from collections import Counter
from dotenv import dotenv_values

# python-dotenv logs "could not parse statement" for every malformed line; silence it
logging.getLogger("dotenv.main").setLevel(logging.CRITICAL)

ALPHABET = ["K", "=", " ", "#", "'", '"', "$", "{", "}", "\\", "v", "1", "export "]

NODE_BATCH = """
const d = require('dotenv');
const lines = JSON.parse(require('fs').readFileSync(0, 'utf8'));
console.log(JSON.stringify(lines.map(l => d.parse(Buffer.from(l)))));
"""


def gen():
    n = random.randint(2, 10)
    return "".join(random.choice(ALPHABET) for _ in range(n)) + "\n"


def py(line):
    try:
        return dict(dotenv_values(stream=io.StringIO(line)))
    except Exception as e:
        return {"error": type(e).__name__}


def node_all(lines):
    p = subprocess.run(["node", "-e", NODE_BATCH], input=json.dumps(lines),
                       capture_output=True, text=True, check=True)
    return json.loads(p.stdout)


def features(line):
    f = []
    if "#" in line: f.append("#")
    if "$" in line or "{" in line: f.append("$ or {")
    if '"' in line: f.append('"')
    if "'" in line: f.append("'")
    if "\\" in line: f.append("backslash")
    if line.startswith("export ") or " export " in line: f.append("export")
    return f or ["(none of these)"]


def main():
    seed = int(sys.argv[1]) if len(sys.argv) > 1 else 0
    count = int(sys.argv[2]) if len(sys.argv) > 2 else 2000
    random.seed(seed)
    lines = sorted({gen() for _ in range(count)})        # unique lines only
    node = node_all(lines)

    found = {}
    for line, n in zip(lines, node):
        p = py(line)
        if json.dumps(p, sort_keys=True) != json.dumps(n, sort_keys=True):
            found[line] = {"python": p, "node": n}

    # Lines with no "=" are a class of their own: python-dotenv reports the bare name
    # with value None, Node ignores the line. Count them, then look at everything else.
    no_eq = {l: r for l, r in found.items() if "=" not in l}
    found = {l: r for l, r in found.items() if "=" in l}

    print(f"seed={seed}: {len(lines)} unique lines, {len(no_eq) + len(found)} disagree")
    print(f"  {len(no_eq)} have no '=' (python-dotenv: name -> None, Node: line ignored)")
    print(f"  {len(found)} contain '=' and still disagree")
    print("\nShortest disagreements among lines with '=':")
    for line, res in sorted(found.items(), key=lambda kv: (len(kv[0]), kv[0]))[:25]:
        print(f"  {line!r:22} python={json.dumps(res['python'])}  node={json.dumps(res['node'])}")

    c = Counter(ft for line in found for ft in features(line))
    print("\nDisagreeing lines with '=' that contain each feature:")
    for ft, k in c.most_common():
        print(f"  {ft:16} {k:5}  ({100 * k / len(found):.0f}%)")


if __name__ == "__main__":
    main()

A note on the first version of the output

The first version of this fuzzer printed one python-dotenv could not parse statement starting at line 1 warning per malformed line, and with random input that is hundreds of lines. They flood the terminal and push the actual results off the screen. That is why the version above silences the dotenv.main logger. The warnings are useful when you hit one in real life (they are the only signal python-dotenv gives), and useless as a fuzzing log.

Results

Run with the default seed and 2,000 draws:

(.venv) user@host:~/dotenv-difference$ python3 fuzz.py
seed=0: 1902 unique lines, 965 disagree
  628 have no '=' (python-dotenv: name -> None, Node: line ignored)
  337 contain '=' and still disagree

Shortest disagreements among lines with '=':
  '"=\n'                 python={"\"": ""}  node={}
  '$=\n'                 python={"$": ""}  node={}
  '}=\n'                 python={"}": ""}  node={}
  '$=1\n'                python={"$": "1"}  node={}
  'K#=\n'                python={"K": null}  node={}
  'K="\n'                python={}  node={"K": "\""}
  'K=#\n'                python={"K": "#"}  node={"K": ""}
  ' 1=#\n'               python={"1": "#"}  node={"1": ""}
  '}==\n'                python={"}": "="}  node={}
  ...

Disagreeing lines with '=' that contain each feature:
  $ or {               215  (64%)
  backslash            147  (44%)
  "                    140  (42%)
  #                    134  (40%)
  '                    105  (31%)
  export                45  (13%)
  (none of these)       10  (3%)

(Output shortened here. The feature counts overlap, because a line can contain several of these characters.)

Scaling up to 20,000 draws (17,047 unique lines) gives the full picture. Nearly half of all random one-line files are read differently:

Class	Lines	Share of disagreements
No `=` in the line	5,183	63%
Name grammar (python-dotenv accepts a name Node rejects, or the reverse)	2,755	33%
Value differences with a valid name	320	4%
Total disagreeing	8,258	48% of 17,047

Exact counts depend on the random generator and on the Python version, so rerun with the same seed to reproduce them. The examples below come from this larger run.

What the fuzzer found that nobody listed

None of these were in the corpus.

1. Lines without =. Given a line that is just K, python-dotenv returns {"K": null}: the name exists, with no value. Node ignores the line. For tools that count keys, this is a different answer to "is K defined?" before any value is even involved. It is the largest class by far.

2. Name grammar. Node only accepts names made of letters, digits, _, . and -. python-dotenv accepts nearly anything before the =, so $=1, }= and "= define variables in Python and are skipped silently in Node. Real names rarely look like this, but it shows that the two readers don't share a definition of a variable name, and a typo can be a valid name in one and a dropped line in the other. A variant: K#= gives python-dotenv a key K with no value (the #= appears to be read as a comment), and Node ignores it.

3. export =v. Node treats export as the variable name and sets {"export": "v"}. python-dotenv rejects the line and sets nothing. A stray space between export and the name turns a statement into a variable called export in one reader.

4. Unterminated single quotes. The corpus had the double-quote case. The fuzzer shows the single-quote case behaves the same way: K=' is dropped entirely by python-dotenv, and Node keeps the stray ' as the value.

5. # in new places. The corpus covered # at the start of a value and in the middle. The fuzzer adds the cases where it follows a quote or a brace, such as 1=v# (python keeps v#, Node keeps v) and v="#= (python drops the line, Node keeps "). Same root cause as observation 3, more places for it to bite.

The disagreements that do involve a valid name and a value (the 4% row) are almost all variations on two themes already in the table: # handling and unterminated quotes. That is a good sign for the hand-written corpus: it covers the common axes, and the fuzzer's extra finds are in the name grammar and the no-= lines.

Tips for making a fuzzer useful

Separate the noise. One trivial class (here, no =) can be two thirds of the output. Split it out so the interesting disagreements are visible.
Show the shortest cases first. A three-character disagreement is a lesson. A twelve-character one is noise.
Bias the alphabet toward the divergence axes you already know: #, quotes, spaces, $, =. Uniform random bytes mostly produce garbage that every reader rejects the same way.
Keep a seed so a finding is reproducible, and promote interesting finds into the fixed corpus.
Add readers incrementally. Every new reader multiplies the disagreements, which is the point.

Limits of this fuzzer

It compares two readers, and one line at a time. It doesn't cover bash, which executes its input (a random line could run a command, so don't fuzz it on a machine you care about), and it doesn't cover the Docker parsers, which need a container per input and are slow. It also can't generate multi-line values or long files. Extending it to Compose and docker run --env-file would likely find more, since those two parsers disagree on 9 of 16 fixtures.

9. Run it in CI

The harness earns its keep when it runs automatically:

# .github/workflows/dotenv-matrix.yml
name: dotenv-matrix
on:
  pull_request:
    paths: ["cases/**", "readers/**", "make_cases.py", "compare.py",
            "run_all.sh", "requirements.txt", "package-lock.json"]

jobs:
  matrix:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - uses: actions/setup-node@v4
        with: { node-version: 22 }
      - run: pip install -r requirements.txt   # python-dotenv==1.2.2, pinned
      - run: npm ci
      - run: ./run_all.sh
      - name: Fail if the committed results are stale
        run: git diff --exit-code results/

Pin every reader to an exact version, including the Docker image tag. When Dependabot or Renovate bumps one, the diff on results/ shows precisely which cells changed. That turns "did the new release change how it parses my config?" from a surprise into a reviewable diff.

Two things to check the first time it runs:

The results must be deterministic. The Docker adapter filters Compose's progress lines because they contain a random container name. If git diff fails on an unchanged commit, look at the stderr fields in results/compose.json first.
Reproduce locally with the same versions. The matrix in this post was produced with Python 3.14, and CI above uses 3.12. The adapters don't depend on the Python version, but if a diff appears, check that first.

10. Takeaways

Measure, don't recall. Even a careful write-up got python-dotenv wrong until it was actually run.
Treat ambiguity as something to detect. You can't make the readers agree, but you can flag lines on which they would disagree, such as KEY=<spaces>#..., an unquoted # in a value, export or spaces around =.
Be strict in what you write. When a tool emits a .env or uploads a value, quote anything containing #, spaces or $. Quoted values are where most dialects agree, with the exception of docker run --env-file, which keeps the quotes.
Keep secrets and comments apart. Put the comment on its own line, or quote the value. API_KEY= on one line and # fill this in later above it is read the same way by every reader we tested.
Check which reader your deployment really uses. Compose, docker run --env-file, your app's library and a developer's source .env are four different parsers.
Fuzz what you can't list. The corpus covered the cases we knew about. The fuzzer added the name grammar, lines without =, and export =v.

The full matrix, the fixtures, the adapters and the fuzzer are small enough to adopt in an afternoon. Fill in the godotenv, dotenvy and Ruby columns for your own stack, and you'll know exactly where your .env files are ambiguous.

The code and every result file are in the companion repo: urwithajit9/dotenv-difference.

The Railway Env-Var Breach: What Happened and How to Be Ready Next Time

Ajit Kumar — Thu, 22 Oct 2026 07:03:01 +0000

On October 21, 2026, Railway told every customer to rotate every environment variable they had stored on the platform. If you deploy anywhere that injects env vars at runtime, the same thing can happen to you, and the only thing you control is how fast you recover.

TL;DR

Railway says an attacker reached encrypted environment variables and, for some, the decrypted values.
No tool on the developer side could have prevented a platform-side breach like this. Good tooling decides whether rotation takes 20 minutes or two days. Managing env files before deployment, with one versioned source the whole team pulls from, is what makes the difference.
Below: what happened, a post-incident rotation runbook, how to prepare before the next one, and an honest comparison of the tools that help.

Disclosure: I build evnx, an open-source .env security CLI with an end-to-end encrypted cloud vault. I use it as the worked example here, but I also cover where other tools are the better fit. My team does not run anything on Railway, so this is written from the outside.

What happened

Railway detected unusual activity in the systems that store customer environment variables on the morning of October 21. Its official disclosure says an unauthorized party may have accessed encrypted variables and, for some of them, obtained the decrypted contents. Those variables typically hold API keys, database passwords and private registry credentials.

Key facts from the disclosure and follow-up coverage:

Who was hit. Railway's initial analysis suggests a subset of users, but it asked all users to rotate as a precaution. It will email confirmed affected accounts directly.
Why. Railway believes this is part of a series of attacks on platform companies aimed at compromising AI inference providers. It notified those upstream providers so they could revoke keys proactively.
Deployments paused. All builds were paused during the investigation, so customers could rotate a key at the provider but could not ship the new value. Enterprise builds came back first, on October 22 at 03:32 UTC, with other plans restored in stages.
Follow-up guides. Railway published guides on checking for anomalous GCP access and AWS access.
Postmortem. Railway has not said how the attacker got in or how many accounts were affected. A formal postmortem is expected after service is restored.

Scale matters here: Railway said in January 2026 that more than 2 million developers build on it. Follow live updates on status.railway.com.

The paused deployments are the detail most write-ups skip. They exposed a hard trade-off: revoke a leaked key now and take an outage, or leave it live until you can deploy. Teams that knew exactly which secrets lived where could make that call per secret. Teams that didn't had to guess.

Why this kind of breach is hard to defend against

A platform that injects environment variables into your app at runtime has to be able to decrypt them. Encryption at rest protects against a stolen disk. It does not help when the attacker reaches the system that holds the keys. That is a property of the env-var model, not a Railway-specific failure.

It also creates the secret-zero problem. Say you move secrets into an external vault and have your app fetch them at boot. The app still needs a credential to reach that vault, and that credential is... an environment variable on the same platform. If the platform leaks it, the attacker can read the vault too. You have moved the problem, not removed it.

So the honest goal is not "make a platform breach impossible". It is:

Small blast radius: each leaked credential reaches as little as possible.
Fast, complete rotation: you know every secret that lived on the platform and can replace them all quickly.
Evidence: you can show what was rotated, when and by whom, for customers, auditors and your own postmortem.

Reaction time: why env work done before deployment pays off

When a platform says "rotate everything", revoking keys is the easy part. What eats the hours is everything around it: which secrets were where, how the team gets the new values, whether anyone missed one, and what broke. evnx is built for the stage before deployment, so most of those answers already exist when the incident starts. Each question maps to one command.

Question mid-incident	Without preparation	With evnx
Which secrets lived on the platform?	Click through dashboards, ask around	`evnx vault list`, plus the `secret = true` entries in `.evnx.toml`
How does the team get the new values?	Keys pasted into Slack or DMs, creating a new leak	One `evnx cloud push`; teammates run `evnx cloud pull`
Does everyone know a value changed?	Message every person	`evnx cloud watch` notifies when someone else pushes
Two people rotating at once?	Last write silently wins	The second push is refused with a conflict
Did the new `.env` drop or break anything?	Found out from a failed deploy	`evnx validate`, `evnx diff --fail-on missing`
A rotated key broke production?	Hunt for the old value	`evnx cloud pull --version <n>`
Did anyone read the vault after the leak?	Unknown	`evnx cloud audit`
A teammate leaves, or a laptop is lost?	Re-share everything by hand	`evnx vault revoke` re-keys the vault
Getting values back onto the platform?	Retype dozens of variables	`evnx migrate --to railway --print-only`
Did a key slip into git during the rush?	Hope	`evnx scan`, enforced by pre-commit hooks

Team rotation in practice. Share the vault once, before anything goes wrong:

evnx vault share api/production --with teammate@example.com --role developer
evnx vault members api/production

Then, on incident day, one person rotates and everyone else follows without a single secret crossing chat:

# The person rotating
evnx validate --env-name production && evnx cloud push

# Everyone else
evnx cloud watch --vault api/production   # tells you a new version landed
evnx cloud pull                           # fetch and decrypt the new values locally

cloud watch only notifies; it never pulls on its own, so nobody's local edits get overwritten. The team works from one versioned source instead of a chain of messages, and that is where most of the saved time comes from.

Post-incident: a rotation runbook with evnx

Here is how the recovery looks if your secrets live in evnx vaults, one vault per service and environment. Commands are from evnx 0.10.15; every flag is in the command reference.

1. Get the list of what to rotate. The vault listing shows which key names each vault holds (DATABASE_URL, STRIPE_KEY and so on). The server stores names in the clear on purpose so a listing is useful, and that listing becomes your rotation checklist. No digging through dashboards or old Slack threads.

2. Revoke at the source, by risk. Database credentials, cloud and IAM keys, payment keys, AI provider keys and registry tokens go first, even if it means downtime while deploys are paused. Read-only or low-value tokens can wait for builds to return.

3. Update locally and validate before anything ships.

evnx validate --env-name production                  # placeholders, weak secrets, duplicate keys, DEBUG="False"
evnx diff --env-name production --fail-on missing    # did the rotation drop a variable the template declares?

A panicked rotation is exactly when someone pastes CHANGE_ME into production or forgets a variable.

4. Store the new values as a new version, and check who touched the vault.

evnx cloud push                            # encrypts locally, uploads ciphertext as a new version
evnx cloud history                         # what changed, when, and who pushed it
evnx cloud audit --vault api/production    # who pushed, pulled, shared or revoked

If a CI credential for the vault lived on the breached platform, cloud audit shows whether anyone pulled after the leak. Every push is a new version and nothing is overwritten. If a rotated key breaks something, evnx cloud pull --version <n> restores the previous one while you investigate.

5. Load the new values into the platform when deploys resume.

evnx migrate --env-name production --to railway \
  --railway-project <project> --railway-environment production --print-only

evnx prints a single railway variable set command for you to read and run; it contacts nothing itself. The same works for Vercel, Heroku, Doppler, Infisical and the AWS, GCP and Azure secret stores.

6. Make sure the old or new keys didn't leak somewhere else.

evnx scan --severity high
evnx scan --format sarif > scan.sarif

It detects AWS (including STS), GitHub, Slack, Google, Stripe, OpenAI and Anthropic keys, JWTs, private keys and passwords in connection strings. It reads every .env variant it finds, such as .env.production. In a git repo it reports what git would commit, so your gitignored .env doesn't fail the run, but a key pasted into a tracked script does.

7. Keep the evidence. cloud history and cloud audit give you a dated record of every rotation and every access. If you're working toward SOC 2, ISO 27001 or Korea's ISMS-P, that is the evidence an auditor asks for after an incident.

Then do the checks Railway recommends: review usage and billing for every AI provider key, grouped per key, and look at database and cloud logs for access you don't recognize.

Pre-incident: set up today, recover fast tomorrow

The runbook above only works if the groundwork is already in place. This is what to set up now.

A source of truth that isn't the platform. If your hosting dashboard is the only place a secret lives, a platform incident leaves you rotating blind. Keep the canonical copy in a vault the platform can't read:

brew install urwithajit9/evnx/evnx      # or: npm install -g @evnx/cli, pipx install evnx
evnx commands                           # should report: built with net, migrate, backup, cloud
evnx vault create api --env production
evnx cloud link api/production
evnx cloud push

The prebuilt binaries include the cloud commands; PyPI has carried them since 0.10.0. Encryption happens on your machine (AES-256-GCM), and the server stores only ciphertext, so a database dump of the evnx server yields no plaintext. Cloud sync is still in beta; the next section says exactly what that means.

One vault per service and environment. That gives clean boundaries: when one platform is compromised, you know exactly which vaults were deployed there.

Write down which variables are secrets.

evnx spec init     # infers a contract for every variable into .evnx.toml

Mark a credential secret = true and scan reports it even when its name gives nothing away. validate enforces which variables each environment requires. That contract is your rotation checklist, written before you need it.

Keep plaintext off disk.

evnx cloud run --vault api/production -- ./deploy.sh

The values are decrypted in memory and handed to the child process: no .env file, nothing in shell history, nothing visible in ps.

Scoped, expiring tokens for CI.

evnx auth token create ci --scope read --vault api/production --expires-in-days 90

A leaked CI token reaches one vault, read-only, for at most 90 days.

Clean offboarding. evnx vault revoke re-keys the vault, so a removed member cannot open versions pushed after they left.

Stop the other leak path: git. Platform breaches get the headlines, but committed secrets are still the most common leak. Run evnx as pre-commit hooks:

repos:
  - repo: https://github.com/urwithajit9/evnx
    rev: v0.10.15          # pin 0.10.0 or newer
    hooks:
      - id: evnx-scan        # blocks commits containing secrets
      - id: evnx-validate    # blocks a misconfigured .env
      - id: evnx-diff        # blocks if .env lacks a key the template declares
      - id: evnx-sync        # blocks if the template lacks a variable from .env
      - id: evnx-scan-push   # strict scan on push

Add evnx validate --format github-actions and evnx scan --format sarif to CI, so findings show up in GitHub's Security tab. evnx doctor --fix makes sure every env file is gitignored, untracked and readable only by you.

Turn on 2FA, and do a rotation drill. evnx auth totp enable protects the vault account. Then pick a quiet afternoon and rotate one non-critical service end to end, using the runbook above. The first rotation is always the slowest; do it before an incident forces you to.

What evnx would not have done

I'd rather say this plainly than have you find it out mid-incident.

It would not have stopped this breach. If your app reads secrets from Railway env vars, Railway holds them in usable form. evnx protects its own copy, not the platform's.
Pulling from evnx at boot doesn't escape secret zero. A pipeline pull, or cloud run, needs an EVNX_TOKEN and the master password. Store both as env vars on a breached platform and the attacker can open the vault. The token only authenticates; the password is what decrypts, and it is the more sensitive of the two. Scoping limits what the token can reach. It is not a substitute for rotation.
Cloud sync is in beta, and it does not yet survive a server takeover. What is already stored stays closed: a database dump or stolen backup yields no plaintext. But an attacker with write access to the evnx server could present a vault key of their choosing and read what you push next. Signed key wraps and locally pinned identities would fix that, and they aren't built yet. Our own README used to say the vault was safe "even after a breach". That was wrong, and we corrected it. Until this is fixed, don't keep secrets there that you couldn't rotate.
It does not rotate keys for you. You still revoke at each provider by hand. Infisical and Doppler can automate rotation for some credential types (below).
It does not verify whether a leaked key is live. TruffleHog does that.
It doesn't call Railway's API. evnx migrate --to railway prints the command for you to run; evnx convert --to railway exists but is marked experimental.

Other tools worth knowing

No single tool covers this whole problem. These are the ones I'd reach for, and what each does best.

Infisical: an open-source secrets platform. It offers syncing to GitHub Actions, GitLab CI, Vercel, AWS and Kubernetes, dynamic secrets, and automated rotation for PostgreSQL, MySQL, AWS IAM and more. It can be self-hosted, including on Railway itself (Postgres + Redis + the app). Best for teams that want automated rotation and short-lived database credentials, and can run a heavier service.

Doppler: hosted secrets management with a native Railway integration. It syncs a Doppler config to a Railway project and environment continuously and can redeploy on change. Railway's own docs describe syncing from Doppler, Infisical or Vault as a supported pattern. Best for teams who want rotation to propagate to Railway automatically and are happy with a hosted SaaS. One caveat: the sync needs a Railway API token, which is a powerful credential in its own right.

dotenvx: encrypted .env files in git. It encrypts values with a public key so the encrypted file can be committed, while the private key in .env.keys stays out of git. Best for small teams who want secrets versioned alongside code. On a platform it still needs the private key as an env var, so the same secret-zero caveat applies.

Gitleaks and TruffleHog: secret scanners. Gitleaks is fast, regex-based and MIT-licensed, which makes it great as a pre-commit hook. TruffleHog's standout feature is verification: it tests whether a detected credential is still live. After an incident, that tells you which keys are still dangerous.

Railway's native options. Railway itself offers service, shared, reference and sealed variables. Use them for least privilege inside the platform, but remember they live on the platform that was breached.

Where evnx fits, and where it excels

evnx is built for the developer side of secrets: the .env on your laptop, onboarding, pre-commit checks, CI, and the path from local config to whatever platform you deploy to.

Tool	Best at	Weak spot in a platform breach
evnx	Whole developer workflow in one CLI, plus an encrypted vault with audit	No automated rotation or live-key verification; cloud sync in beta
Infisical	Automated rotation, dynamic DB credentials, broad syncs	Heavier service to run; its runtime token is secret zero
Doppler	Continuous sync into Railway, redeploy on change	Hosted SaaS; sync needs a powerful Railway API token
dotenvx	Encrypted .env committed to git	Private key on the platform is secret zero
Gitleaks	Fast pre-commit scanning	Finds secrets, doesn't verify them
TruffleHog	Verifying which leaked keys are live	Scanner only, no storage or rotation

Where evnx stands out:

Encrypted on your machine. The server stores ciphertext and never holds a usable key, so a stolen database is useless. Team sharing wraps vault keys with hybrid X25519 + ML-KEM-768, and revoking a member re-keys the vault.
One tool for the whole recovery. Validate, diff, scan, version, audit and roll back, then hand the values to 14 conversion targets or print migration commands for Railway, Vercel, Heroku, Doppler, Infisical, AWS, GCP and Azure. In the middle of an incident, one tool beats four.
It catches the mistakes rotation creates. Placeholder values, the DEBUG="False" truthy trap, duplicate or missing variables and .env / .env.example drift are what break production right after a hurried rotation.
Plaintext stays off disk. cloud run injects secrets straight into a process, with no file and nothing in ps.
Clear answers in CI. Exit code 0 means nothing found, 1 means a problem was found, and 2 means evnx could not finish. "I found nothing" and "I couldn't look" are never confused.
Lightweight and portable. A single Rust binary, MIT-licensed, installable from Homebrew, npm, pipx, Cargo, Scoop or winget. The server is open source and self-hostable.

If you need automated rotation for databases today, use Infisical or Doppler. evnx can still sit alongside them on the developer side.

Get started with evnx

Installing takes one line, and scanning or validating needs no account. Pick a channel:

curl -fsSL https://evnx.dev/install.sh | bash   # macOS and Linux, prebuilt binary
brew install urwithajit9/evnx/evnx              # Homebrew
npm install -g @evnx/cli                        # npm
pipx install evnx                               # PyPI
cargo install evnx --all-features               # from source, every feature
winget install urwithajit9.evnx                 # Windows

Then run a first check on a project you already have:

evnx doctor    # env files gitignored, untracked, not world-readable
evnx scan      # secrets in any file git would commit

For encrypted vaults and team sharing, create a free account. The free tier includes 3 vaults, 5 versions each and 2 API tokens, with no card required.

evnx.dev: the project site
Quick start: your first audit in five minutes
Installation guide: all nine install channels
Command reference: every command, flag and exit code
Cloud architecture: how the encryption works, including what it does not protect against
GitHub: source, issues and discussions

Your checklist

If you're on Railway right now:

[ ] Revoke database, cloud, payment, AI provider and registry credentials at each provider
[ ] Check AI provider usage and billing per key, plus database and cloud access logs
[ ] Follow Railway's GCP and AWS guides if you use either
[ ] Redeploy with new values once builds return, then confirm the old values fail
[ ] Watch your inbox for Railway's direct notice to affected accounts

Wherever you deploy, before the next incident:

[ ] Keep a source of truth for secrets that your hosting platform can't read
[ ] Split secrets per service and environment, and know which platform each set is deployed to
[ ] Use scoped, expiring tokens in CI
[ ] Scan and validate in pre-commit and CI
[ ] Run one rotation drill end to end

Platform breaches will keep happening, especially while AI inference keys are worth stealing. You can't stop the next one. You can make sure it costs you an afternoon, not a week.

Try evnx with the install line above, read the docs, and star or open an issue on GitHub. I'd love feedback from anyone who rotated through this incident: what slowed you down?

Sources

From AWS Key Leak to evnx: The Origin Story of a Developer's Safety Net

Ajit Kumar — Wed, 01 Apr 2026 08:30:06 +0000

How an AWS key leak, years of .env frustrations, and a poorly formatted LLM prompt accidentally built a developer tool.

Part 1: The Accident — An AWS Key Leak That Changed Everything

Every developer has that story. The one they hesitate to tell in interviews, wince at in retrospectives, and quietly reference whenever someone asks why security tooling matters. This is mine.

About a year ago, I was deep in a refactor of an Apache Airflow project at work — my first Airflow project, born from early-career learning and grown organically into a system with 10+ DAGs and a sprawling network of helper Python files. The project had been running for over a year. Scrapers, ETL tasks, one-off experiments that graduated to production — each added in its own folder, each with its own .env file.

At the end of a long day of refactoring, I just wanted to save my work. No ceremony. A quick git add ., a commit, and a push. The project didn't have pre-commit hooks at the time. Within minutes, things started breaking. Services went down. I went digging through backend code for bugs I'd introduced that day, and found nothing. Then I opened my email.

GitHub. AWS. Key revocation notices.

I had accidentally committed one of the nested .env files — one that contained AWS credentials and keys for several backend services. Everything connected to those keys was now broken. I had to explain the incident to my supervisor. I had to trace the blast radius. I had to rotate credentials and restore services one by one.

It was embarrassing. More than that, it was a moment that stuck with me at a deeper level than any code review comment ever had.

The immediate fix was pragmatic: I added pre-commit with secret scanning and moved on. But the underlying problem — the real one — quietly settled into the back of my mind, waiting.

Part 2: Years of `.env` Pain Across Projects

The AWS incident was acute, but the chronic pain had been building for years. Working across 7–8 projects at my previous job, plus personal and consulting work, I kept running into the same categories of friction around .env management — just in different flavors each time.

The .env vs .env.example drift problem. The idea is simple: commit a .env.example with all the keys but no real values, so other developers know what they need. In practice, as a project matures over months, keys get added to .env and never make it into .env.example. New team members clone the repo, fill out the example file, and then hit cryptic runtime errors because the example was out of date. What seems like a one-line discipline problem compounds into real onboarding pain.

Sharing and syncing across a distributed team. Software is a team sport, and every team member needs a working .env.local, .env.staging, or environment-specific variant. Add remote team members to the mix — especially part-timers with high turnover — and you have a constant cycle of sharing, revoking, rotating, and re-syncing credentials. For me, managing this for a team where people were frequently entering and leaving was one of the most draining non-coding tasks in the development workflow.

The wild west of .env formats. There's no formal specification for .env files. Every tool interprets them slightly differently. python-dotenv will happily parse quoted values like KEY="value" and strip the quotes for you. Docker's --env-file parser is stricter and will choke on the same syntax when building images. And then there are the migrations: for one project, I needed to move 30+ variables from a .env file into AWS Secrets Manager, which required a JSON format. Another project needed those same variables pushed into GitHub Actions secrets. Each migration meant careful, highly concentrated copy-paste — the kind where one missed variable or one formatting mistake could mean a broken deployment or a security gap. Nobody should be doing this manually.

All of it — the drift, the syncing, the format fragmentation — was death by a thousand cuts. Manageable, individually. Maddening, collectively.

Part 3: The Poorly Written Prompt That Started It All

Here's where the story takes an unexpected turn.

Sometime after the AWS incident, sitting with all of these accumulated frustrations about .env files, I did what many developers do: I opened an LLM and started typing. I wanted a comprehensive resource — a tutorial covering what .env files are, how to use them correctly, their limitations, and how different languages and ecosystems handle them.

My mistake: I forgot to specify the output format. I assumed it would come back as Markdown.

The LLM returned a beautifully structured HTML document.

I read through it. It was actually quite good. And in a moment of "well, it's just one file," I decided to host it. Bought a domain. Deployed the single HTML page. That became dotenv.space.

I added Google Analytics and Plausible, mostly out of curiosity. Within days, organic search traffic was arriving. People were landing on the page. The data made one thing clear: there was genuine demand for better .env tooling, not just documentation.

That weekend, I decided to dust off my Rust skills — about a year of on-and-off learning — and build a CLI that could actually solve the problems I had been living with. The first working version of what would become evnx came together in three to four days, with LLM assistance bridging the gaps in my Rust knowledge. I iterated on it, fixed bugs, used it on real projects, then did something that revealed the real state of the code: I read it line by line.

Even as a Rust beginner, I could see the issues. The structure wasn't right. The abstractions were leaking. So I started over — not from scratch, but from understanding. The second iteration was better. The third was something I felt good enough to ship.

Part 4: From `dotenv-cli` to `evnx` — A Name Born From Distribution

As the CLI matured through iterations, I turned my attention to distribution — a completely new problem space for me. I wanted the tool to be accessible everywhere developers already were: Cargo for Rust users, pip for Python users, npm for JavaScript developers, Homebrew for macOS, Scoop for Windows, winget for Windows package management, and GitHub Actions for CI/CD pipelines.

Each distribution channel has its own conventions, its own packaging requirements, its own review process. Working through all of them was an education in itself.

When it came to naming, I ran into a constraint: the name I originally had in mind — envX — was already taken across distribution channels. I needed something close, something that captured the spirit of the project.

The solution: evnx. Phonetically adjacent, practically distinct, and backronymed into something meaningful — Environment Variables Next Generation.

The name felt right. It acknowledged the lineage while staking out new territory.

Part 5: Where Things Stand Now

I'm currently in the gap between my last role and whatever comes next — a rare window of time that I've chosen to spend building evnx into something genuinely useful for other developers.

I use it myself. It solves the real problems that led to its creation: validating .env files, keeping .env.example in sync, scanning for accidentally committed secrets, converting between formats (JSON, YAML, Kubernetes, Docker, GitHub Actions, and more), and integrating into CI/CD pipelines via GitHub Actions.

The origin wasn't glamorous. It started with embarrassment, accumulated through years of low-grade frustration, and got catalyzed by a formatting oversight in an LLM prompt. But that's how a lot of useful tools get built — not from grand vision, but from lived pain that finally crosses the threshold of tolerance.

If any of this sounds familiar, I'd love for you to try evnx, file issues, suggest features, and help shape what it becomes.

Get Started with evnx

🌐 Official Website

evnx.dev — Documentation, installation guides, and full feature reference. Start here if you're new to evnx.

📦 Main Repository

github.com/urwithajit9/evnx — The core CLI source code. Open issues, submit pull requests, and follow development here.

Status:
✅ CI Passing •
📦 Latest Release •
🦀 crates.io •
🐍 PyPI •
📦 npm •
📄 MIT License •
🔧 GitHub Action

🧪 Test Repository

github.com/urwithajit9/evnx-test — Sample .env files, test fixtures, and integration test cases. Useful if you want to explore evnx commands against realistic inputs before running it on your own project.

⚙️ GitHub Actions Integration

github.com/urwithajit9/evnx-action — The official evnx GitHub Action. Add .env validation and secret scanning directly to your CI/CD pipeline with a few lines of YAML. Prevent leaks before they reach your repository.

🍺 Homebrew Tap (macOS / Linux)

github.com/urwithajit9/homebrew-evnx — Install evnx on macOS or Linux via Homebrew:

brew tap urwithajit9/evnx
brew install evnx

🪟 Scoop Bucket (Windows)

github.com/urwithajit9/scoop-evnx — Install evnx on Windows via Scoop:

scoop bucket add evnx https://github.com/urwithajit9/scoop-evnx
scoop install evnx

evnx is open source under the MIT License. Contributions, feedback, and stars are all deeply appreciated.

Python 3 & Security: A Deep-Dive Tutorial

Ajit Kumar — Fri, 13 Mar 2026 05:40:09 +0000

Assessment Focus: Python 3 built-ins · Functional tools · Modern Python 3.13 · Security fundamentals
Level: Beginner → Advanced — structured for systematic reading and practice
Time to complete: ~90 minutes reading + practice

Python 3.13 — What Changed and Why It Matters
Input & Safe Data Entry
Core Iteration Built-ins — zip, enumerate, map, filter
Aggregation Built-ins — sum, any, all, min, max
Sorting — sorted, sort, key functions
Lambda Functions & Functional Style
Comprehensions — list, dict, set, generator
Exception Handling — try, except, else, finally
Custom Exceptions & Exception Chaining
Context Managers & Resource Safety
String Security — Formatting, Sanitisation, Validation
Injection Attacks — SQL, Command, Path Traversal
Password & Secret Handling
Cryptography Fundamentals in Python
Authentication — JWT, Sessions, Tokens
Timing Attacks & Side Channels
Dangerous Python Patterns to Avoid
Practice Problems with Full Solutions
Assessment Cheat Sheet

1. Python 3.13 — What Changed and Why It Matters

Python 3.13 (released October 2024) introduced several changes that directly affect how you write safe, idiomatic code. These are the ones most likely to appear in an assessment testing "modern Python" knowledge.

1.1 The New Interactive Interpreter (REPL)

The REPL was completely rewritten using PyPy's pyrepl library:

# Python 3.13 REPL features:
# ✦ Multi-line editing — navigate up/down within a block
# ✦ Syntax highlighting in the terminal
# ✦ Exit with 'exit' (no parentheses needed, though exit() still works)
# ✦ F1 brings up interactive help browser
# ✦ F2 strips output/prompts for clean copying
# ✦ F3 toggles "paste mode" — prevents auto-indentation issues

# How to check your Python version in code:
import sys
print(sys.version)          # '3.13.0 (main, Oct 7 2024, ...)'
print(sys.version_info)     # sys.version_info(major=3, minor=13, ...)
print(sys.version_info >= (3, 13))  # True

1.2 input() — The Security-Critical Change ⚠️

This is the most important Python 3.13 change for security. In all Python versions prior to 3.13, input() in certain contexts could be manipulated. Python 3.13 reinforces the guarantee that input() always returns a plain string — never evaluates it.

# In Python 2, input() was equivalent to eval(raw_input())
# This was a critical security vulnerability:
# user types: __import__('os').system('rm -rf /')
# Python 2 would EXECUTE it!

# Python 3 (all versions including 3.13):
user_input = input("Enter a value: ")
# Always returns a str, NEVER evaluates
# Even if user types: 2 + 2
# You get the STRING "2 + 2", not the integer 4

# Python 3.13 makes this explicit in the docs and adds stricter
# safeguards in the REPL — input() in all modes is now truly raw

# ✅ Safe pattern for numeric input
def get_integer(prompt: str) -> int | None:
    raw = input(prompt)           # always a string
    try:
        return int(raw)           # explicit conversion, may raise ValueError
    except ValueError:
        print(f"'{raw}' is not a valid integer")
        return None

# ✅ Safe pattern with validation
def get_positive_number(prompt: str) -> float:
    while True:
        raw = input(prompt).strip()
        try:
            value = float(raw)
            if value <= 0:
                raise ValueError("Must be positive")
            return value
        except ValueError as e:
            print(f"Invalid input: {e}. Try again.")

# ❌ NEVER do this — eval on input is always wrong
result = eval(input("Enter expression: "))   # arbitrary code execution!

1.3 Free-Threaded Mode (Experimental in 3.13)

# Python 3.13 introduces experimental "free-threaded" mode (--disable-gil)
# This allows true parallel threads without the GIL
# Build with: --disable-gil flag, check with:

import sys
print(sys._is_gil_enabled())   # False in free-threaded build, True otherwise

# Security implication: without GIL, concurrent code needs explicit locks
# previously "safe" singleton patterns break without threading.Lock

# Check at runtime so code is forward-compatible:
import threading
_lock = threading.Lock()

def thread_safe_operation():
    with _lock:
        # critical section — safe in both GIL and free-threaded mode
        pass

1.4 Improved Error Messages

Python 3.13 error messages are significantly better. Knowing what they mean speeds up debugging:

# NameError — now suggests did you mean?
pint = 3.14
# print(pitn)  → NameError: name 'pitn' is not defined. Did you mean: 'pint'?

# AttributeError — now suggests similar attributes
my_list = [1, 2, 3]
# my_list.appnd(4)  → AttributeError: 'list' object has no attribute 'appnd'.
#                      Did you mean: 'append'?

# TypeError — clearer messages about argument counts
def greet(name, greeting):
    return f"{greeting}, {name}!"
# greet("Alice")  → TypeError: greet() missing 1 required positional argument: 'greeting'

# Python 3.13 also shows the exact location of the error with a caret (^)
# x = {'a': 1}
# x['b']['c']  → KeyError with helpful traceback pointing to 'b'

1.5 Type Parameter Syntax (PEP 695, stabilised in 3.13)

# Old style (still valid):
from typing import TypeVar
T = TypeVar('T')
def first(lst: list[T]) -> T:
    return lst[0]

# New style — Python 3.12+ (3.13 polishes this):
def first[T](lst: list[T]) -> T:       # T declared inline
    return lst[0]

type Vector = list[float]               # type alias statement
type Matrix[T] = list[list[T]]          # generic type alias

# These are documentation/tooling hints — still not enforced at runtime
# Use mypy or pyright for static type checking

✏️ Quick Check 1

Q: A junior developer writes result = eval(input("Enter formula: ")). What's the exact security risk?

A: This is Remote Code Execution (RCE). A user can type any Python expression and it will be executed with the full permissions of the running process. For example: __import__('os').system('cat /etc/passwd') or __import__('subprocess').call(['rm','-rf','/']). Even on internal tools, this violates the principle of least privilege and creates an insider threat vector.

2. Input & Safe Data Entry

2.1 The Full input() Contract

# input() always:
# 1. Displays the prompt string to stdout (no newline)
# 2. Reads one line from stdin
# 3. Strips the trailing newline
# 4. Returns a str — NEVER evaluates, NEVER converts

name = input("Name: ")         # "Alice\n" → "Alice"
print(type(name))              # <class 'str'>

# Edge cases
empty = input("Press Enter: ") # user just presses Enter → ""
space = input("Enter: ")       # user types "  " → "  " (spaces preserved!)

2.2 Robust Input Validation Patterns

import re
from typing import Callable, TypeVar

T = TypeVar('T')

def validated_input(
    prompt: str,
    converter: Callable[[str], T],
    validator: Callable[[T], bool] | None = None,
    error_msg: str = "Invalid input",
    max_attempts: int = 3
) -> T:
    """Generic validated input — try up to max_attempts times."""
    for attempt in range(1, max_attempts + 1):
        raw = input(prompt).strip()
        try:
            value = converter(raw)
            if validator is not None and not validator(value):
                raise ValueError(error_msg)
            return value
        except (ValueError, TypeError) as e:
            remaining = max_attempts - attempt
            print(f"  Error: {e}. {remaining} attempt(s) remaining.")
    raise RuntimeError(f"Failed after {max_attempts} attempts")

# Usage examples:
age   = validated_input("Age: ", int, lambda x: 0 < x < 150, "Age must be 1–149")
score = validated_input("Score (0–100): ", float, lambda x: 0 <= x <= 100)
email = validated_input(
    "Email: ",
    str,
    lambda s: bool(re.fullmatch(r'[^@\s]+@[^@\s]+\.[^@\s]+', s)),
    "Not a valid email address"
)

2.3 Security-First Input Rules

import html
import unicodedata

# Rule 1: Strip and limit length BEFORE processing
def safe_input(prompt: str, max_length: int = 256) -> str:
    raw = input(prompt)
    stripped = raw.strip()
    if len(stripped) > max_length:
        raise ValueError(f"Input too long (max {max_length} chars)")
    return stripped

# Rule 2: Normalise Unicode to prevent homograph attacks
# "аdmin" (Cyrillic 'а') looks identical to "admin" (Latin 'a')
def normalise_unicode(s: str) -> str:
    return unicodedata.normalize('NFKC', s)  # normalises lookalike chars

print(normalise_unicode("аdmin") == "admin")   # True — detected!

# Rule 3: HTML-escape before rendering in web contexts
user_name = "<script>alert('xss')</script>"
safe_name = html.escape(user_name)
# "&lt;script&gt;alert(&#x27;xss&#x27;)&lt;/script&gt;"

# Rule 4: Whitelist, don't blacklist
def is_safe_username(s: str) -> bool:
    # Allow ONLY alphanumeric + underscore, 3–32 chars
    return bool(re.fullmatch(r'[a-zA-Z0-9_]{3,32}', s))

# Blacklist approach is always incomplete — attackers find bypasses
def is_safe_username_bad(s: str) -> bool:
    return '<' not in s and '>' not in s   # misses %3C, &#60;, etc.

3. Core Iteration Built-ins — zip, enumerate, map, filter

These are the workhorses of Pythonic code. They all return lazy iterators — they don't compute results until you iterate.

3.1 zip — Parallel Iteration

# zip pairs elements from multiple iterables
names  = ["Alice", "Bob", "Carol"]
scores = [95, 87, 91]
grades = ["A", "B+", "A-"]

# Basic zip — stops at shortest iterable
for name, score in zip(names, scores):
    print(f"{name}: {score}")

# zip returns an iterator — materialise with list() if needed
pairs = list(zip(names, scores))    # [("Alice",95), ("Bob",87), ("Carol",91)]

# zip with * operator — unzip (transpose)
pairs = [(1, 'a'), (2, 'b'), (3, 'c')]
nums, letters = zip(*pairs)         # (1,2,3) and ('a','b','c')

# zip_longest — don't stop at shortest (from itertools)
from itertools import zip_longest
a = [1, 2, 3, 4]
b = ['x', 'y']
list(zip_longest(a, b, fillvalue=None))  # [(1,'x'),(2,'y'),(3,None),(4,None)]

# zip three or more iterables
for name, score, grade in zip(names, scores, grades):
    print(f"{name}: {score} ({grade})")

# zip for dict construction
keys   = ['host', 'port', 'db']
values = ['localhost', 5432, 'mydb']
config = dict(zip(keys, values))
# {'host': 'localhost', 'port': 5432, 'db': 'mydb'}

# Security use: zip for parallel processing with validation
def process_user_data(usernames: list, passwords: list) -> list[dict]:
    if len(usernames) != len(passwords):
        raise ValueError("Mismatched username/password counts")  # security check!
    return [
        {'username': u.strip(), 'hashed': hash_password(p)}
        for u, p in zip(usernames, passwords)
        if u.strip()   # skip empty usernames
    ]

3.2 enumerate — Index + Value

fruits = ["apple", "banana", "cherry", "date"]

# Without enumerate — error-prone
for i in range(len(fruits)):
    print(i, fruits[i])    # fragile — easy to have off-by-one errors

# With enumerate — clean and idiomatic
for i, fruit in enumerate(fruits):
    print(i, fruit)         # 0 apple, 1 banana, ...

# Custom start index
for i, fruit in enumerate(fruits, start=1):
    print(f"{i}. {fruit}")  # 1. apple, 2. banana, ...

# enumerate returns (index, value) tuples — useful for list of dicts
records = [{'name': 'Alice', 'score': 95}, {'name': 'Bob', 'score': 87}]
for idx, record in enumerate(records):
    print(f"Record {idx}: {record['name']} scored {record['score']}")

# Security use: enumerate for audit logging with position context
def validate_config_lines(lines: list[str]) -> list[str]:
    errors = []
    for line_num, line in enumerate(lines, start=1):
        line = line.strip()
        if not line or line.startswith('#'):
            continue   # skip empty lines and comments
        if '=' not in line:
            errors.append(f"Line {line_num}: missing '=' in '{line}'")
        key, _, value = line.partition('=')
        if not key.strip():
            errors.append(f"Line {line_num}: empty key")
        if not value.strip():
            errors.append(f"Line {line_num}: empty value for key '{key.strip()}'")
    return errors

3.3 map — Apply a Function to Every Element

# map(function, iterable) → lazy iterator
numbers = [1, 4, 9, 16, 25]
roots   = list(map(lambda x: x ** 0.5, numbers))  # [1.0, 2.0, 3.0, 4.0, 5.0]

# map with a named function
words      = ["  hello  ", "  world  ", "  python  "]
stripped   = list(map(str.strip, words))   # ['hello', 'world', 'python']
uppercased = list(map(str.upper, words))

# map with multiple iterables — zip-like, stops at shortest
a = [1, 2, 3]
b = [10, 20, 30]
sums = list(map(lambda x, y: x + y, a, b))   # [11, 22, 33]

# map vs list comprehension — map is faster for simple named functions,
# comprehensions are faster for inline expressions and more readable
# Generally prefer comprehensions for clarity:
stripped_lc = [w.strip() for w in words]   # preferred

# Security use: sanitise a batch of user inputs
raw_inputs = ["  Alice  ", "<Bob>", "Carol\n", "  dave@example.com  "]
cleaned    = list(map(lambda s: html.escape(s.strip()), raw_inputs))

3.4 filter — Keep Elements Matching a Predicate

import html

numbers = [-3, -1, 0, 1, 2, 3, 4, 5]

# filter(predicate, iterable) — keeps elements where predicate is True
positives = list(filter(lambda x: x > 0, numbers))   # [1,2,3,4,5]
non_zero  = list(filter(None, numbers))               # [nonzero elements] — filter(None,...) removes falsy

# Equivalent comprehension (preferred for clarity)
positives_lc = [x for x in numbers if x > 0]

# filter with a named function
words = ["hello", "", "world", "  ", "python", None, "!"]
# Keep only truthy non-empty strings
valid = list(filter(lambda w: w and w.strip(), words))   # ['hello','world','python','!']

# Chaining filter + map — pipeline style
data    = ["  admin  ", "", "  Bob  ", "   ", "charlie"]
cleaned = list(map(str.upper, filter(lambda s: s.strip(), data)))
# ['ADMIN', 'BOB', 'CHARLIE']

# Security use: filter out suspicious inputs before processing
ALLOWED_PATTERN = re.compile(r'^[a-zA-Z0-9_@.\-]+$')

def filter_safe_inputs(inputs: list[str]) -> list[str]:
    def is_safe(s: str) -> bool:
        s = s.strip()
        return bool(s) and bool(ALLOWED_PATTERN.match(s))
    return list(filter(is_safe, inputs))

suspicious = ["alice", "bob; DROP TABLE users;--", "carol", "../../etc/passwd"]
safe_only  = filter_safe_inputs(suspicious)   # ['alice', 'carol']

✏️ Practice Problem 3

Problem: You have a list of log entries as strings:

"2024-01-15 ERROR user=admin ip=192.168.1.1 msg=login_failed"
"2024-01-15 INFO  user=alice ip=10.0.0.5 msg=login_ok"
"2024-01-15 ERROR user=root ip=203.0.113.5 msg=login_failed"

Using filter, map, and enumerate, extract all ERROR entries, parse each into a dict, and return a list of (line_number, parsed_dict) tuples.

Solution:

logs = [
    "2024-01-15 ERROR user=admin ip=192.168.1.1 msg=login_failed",
    "2024-01-15 INFO  user=alice ip=10.0.0.5 msg=login_ok",
    "2024-01-15 ERROR user=root ip=203.0.113.5 msg=login_failed",
    "2024-01-15 WARN  user=bob ip=172.16.0.1 msg=rate_limited",
]

def parse_log_line(line: str) -> dict:
    parts  = line.split()
    date   = parts[0]
    level  = parts[1].strip()
    fields = {}
    for part in parts[2:]:
        if '=' in part:
            k, _, v = part.partition('=')
            fields[k] = v
    return {'date': date, 'level': level, **fields}

# Step 1: enumerate gives us (line_number, line)
# Step 2: filter keeps only ERROR lines
# Step 3: map parses each into a dict

numbered = list(enumerate(logs, start=1))                           # (linenum, line) tuples

error_entries = list(filter(
    lambda pair: 'ERROR' in pair[1],
    numbered
))   # [(1, "...ERROR..."), (3, "...ERROR...")]

parsed_errors = list(map(
    lambda pair: (pair[0], parse_log_line(pair[1])),
    error_entries
))

for line_num, entry in parsed_errors:
    print(f"Line {line_num}: {entry}")
# Line 1: {'date': '2024-01-15', 'level': 'ERROR', 'user': 'admin', 'ip': '192.168.1.1', 'msg': 'login_failed'}
# Line 3: {'date': '2024-01-15', 'level': 'ERROR', 'user': 'root', 'ip': '203.0.113.5', 'msg': 'login_failed'}

4. Aggregation Built-ins — sum, any, all, min, max

4.1 sum — Flexible Accumulation

# Basic sum — works on any numeric iterable
numbers = [1, 2, 3, 4, 5]
total   = sum(numbers)          # 15

# start parameter — initial value (crucial for non-zero starts)
sum(numbers, start=100)         # 115
sum(numbers, 0)                 # 15 (positional, but named is clearer)

# Summing non-integers — as long as + is defined
from decimal import Decimal
prices  = [Decimal('9.99'), Decimal('4.99'), Decimal('14.99')]
total   = sum(prices, Decimal('0'))   # Decimal('29.97')

# Sum of a generator expression — memory efficient
total_sq = sum(x**2 for x in range(1000))   # no intermediate list!

# Flatten one level with sum + start=[]  (works but is slow — use itertools.chain)
nested  = [[1,2], [3,4], [5,6]]
flat    = sum(nested, start=[])   # [1,2,3,4,5,6]  — but O(n²)! avoid for large lists
# Better:
import itertools
flat    = list(itertools.chain.from_iterable(nested))   # O(n)

# Security use: count failed attempts from events list
events = [
    {'type': 'login_fail', 'user': 'admin'},
    {'type': 'login_ok', 'user': 'alice'},
    {'type': 'login_fail', 'user': 'root'},
    {'type': 'login_fail', 'user': 'root'},
]
fail_count = sum(1 for e in events if e['type'] == 'login_fail')   # 3
# Or: sum(e['type'] == 'login_fail' for e in events)  — bools are ints (True=1, False=0)

4.2 any and all — Logical Reductions

# any(iterable) — True if ANY element is truthy (short-circuits on first True)
# all(iterable) — True if ALL elements are truthy (short-circuits on first False)

flags = [True, False, True, True]
print(any(flags))   # True  — at least one True
print(all(flags))   # False — not ALL are True

# Edge cases — always know these
any([])   # False — no elements to be True
all([])   # True  — vacuously true (no element is False)

# With generator expressions — lazy, short-circuits
numbers = range(1, 1_000_000)
has_even  = any(n % 2 == 0 for n in numbers)   # stops at n=2
all_pos   = all(n > 0 for n in numbers)         # stops at first n ≤ 0 (never, so scans all)

# Security uses — validation pipelines
def validate_password(password: str) -> tuple[bool, list[str]]:
    checks = [
        (len(password) >= 12,              "at least 12 characters"),
        (any(c.isupper() for c in password), "at least one uppercase letter"),
        (any(c.islower() for c in password), "at least one lowercase letter"),
        (any(c.isdigit() for c in password), "at least one digit"),
        (any(c in '!@#$%^&*()_+-=' for c in password), "at least one special character"),
    ]
    failures = [msg for passed, msg in checks if not passed]
    return (len(failures) == 0, failures)

ok, errors = validate_password("MyPass1!")
print(ok, errors)   # False, ['at least 12 characters']

ok, errors = validate_password("MySecurePass1!")
print(ok, errors)   # True, []

# Check multiple permissions at once
def can_access(user: dict, resource: str) -> bool:
    required_permissions = ['authenticated', 'active', f'can_read_{resource}']
    return all(user.get(p, False) for p in required_permissions)

# Validate a list of IP addresses
import ipaddress
def all_ips_valid(ip_list: list[str]) -> bool:
    def is_valid_ip(ip: str) -> bool:
        try:
            ipaddress.ip_address(ip)
            return True
        except ValueError:
            return False
    return all(is_valid_ip(ip) for ip in ip_list)

4.3 min and max — With Custom Keys

students = [
    {'name': 'Alice', 'score': 95, 'age': 20},
    {'name': 'Bob',   'score': 87, 'age': 22},
    {'name': 'Carol', 'score': 91, 'age': 21},
]

# Default min/max on numbers
print(min(3, 1, 4, 1, 5, 9))   # 1
print(max([3, 1, 4, 1, 5, 9])) # 9

# With key function — transforms each element before comparison
top_scorer    = max(students, key=lambda s: s['score'])   # Alice (95)
youngest      = min(students, key=lambda s: s['age'])     # Alice (20)
name_first    = min(students, key=lambda s: s['name'])    # Alice (alphabetical)

# default parameter — avoids ValueError on empty iterable
winner = max([], default=None)    # None instead of ValueError

# Security use: find the most suspicious IP (most failed attempts)
login_attempts = [
    {'ip': '192.168.1.1', 'failures': 3},
    {'ip': '203.0.113.5', 'failures': 47},
    {'ip': '10.0.0.5',    'failures': 1},
]
most_suspicious = max(login_attempts, key=lambda x: x['failures'])
print(most_suspicious['ip'])   # '203.0.113.5'

5. Sorting — sorted, sort, key functions

5.1 sorted vs .sort()

# sorted() — returns a NEW sorted list, works on any iterable
nums = [3, 1, 4, 1, 5, 9, 2, 6]
asc  = sorted(nums)              # [1,1,2,3,4,5,6,9] — nums unchanged
desc = sorted(nums, reverse=True)  # [9,6,5,4,3,2,1,1]

# .sort() — in-place, returns None, only on lists
nums.sort()              # modifies nums in place
nums.sort(reverse=True)  # in-place descending

# GOTCHA: sorted() returns a list even from non-list iterables
sorted_str  = sorted("python")    # ['h','n','o','p','t','y'] — sorted chars
sorted_set  = sorted({3,1,2})     # [1,2,3]
sorted_dict = sorted({'b':2,'a':1})  # ['a','b'] — sorts KEYS only

# Stability — Python's sort is stable (TimSort)
# Elements with equal keys maintain their original relative order
data = [('Bob', 90), ('Alice', 90), ('Carol', 85)]
sorted(data, key=lambda x: x[1])
# [('Carol',85), ('Bob',90), ('Alice',90)] — Bob before Alice preserved!

5.2 The key Parameter — Transformation Before Comparison

# key= receives a function applied to each element ONCE before comparison
# The key values are compared, not the original elements

# Case-insensitive sort
words = ["Banana", "apple", "Cherry", "date"]
sorted(words, key=str.lower)    # ['apple', 'Banana', 'Cherry', 'date']

# Sort by string length, then alphabetically (multi-key sort)
sorted(words, key=lambda w: (len(w), w.lower()))
# ['date', 'apple', 'Banana', 'Cherry']

# Sort dicts by a field
users = [
    {'name': 'Carol', 'last_login': '2024-01-10', 'role': 'admin'},
    {'name': 'Alice', 'last_login': '2024-01-15', 'role': 'user'},
    {'name': 'Bob',   'last_login': '2024-01-12', 'role': 'user'},
]
by_login  = sorted(users, key=lambda u: u['last_login'], reverse=True)
# Most recent login first: Alice → Bob → Carol

# Sort by role then name (admin first, then alphabetical)
by_role   = sorted(users, key=lambda u: (u['role'] != 'admin', u['name']))
# Carol (admin) first, then Alice, Bob

# operator.itemgetter — faster than lambda for dict/tuple access
from operator import itemgetter, attrgetter
by_name   = sorted(users, key=itemgetter('name'))   # faster than lambda u: u['name']

# attrgetter for objects
from dataclasses import dataclass

@dataclass
class User:
    name: str
    score: int
    active: bool

user_list = [User('Bob', 70, True), User('Alice', 90, False), User('Carol', 80, True)]
active_by_score = sorted(
    filter(attrgetter('active'), user_list),   # filter active users
    key=attrgetter('score'),
    reverse=True
)   # [Carol(80), Bob(70)]

5.3 Security Relevance of Sorting

# Sorting for security reporting
def top_n_attackers(events: list[dict], n: int = 10) -> list[dict]:
    """Return top N IPs by number of failed attempts."""
    from collections import Counter
    failure_counts = Counter(
        e['ip'] for e in events if e.get('event') == 'login_failure'
    )
    # Sort by count descending, then IP for determinism
    return sorted(
        [{'ip': ip, 'failures': count} for ip, count in failure_counts.items()],
        key=lambda x: (-x['failures'], x['ip'])
    )[:n]

# Priority queue for processing security alerts
alerts = [
    {'severity': 3, 'msg': 'Port scan detected'},
    {'severity': 1, 'msg': 'Failed login'},
    {'severity': 5, 'msg': 'Root shell spawned'},
    {'severity': 2, 'msg': 'Unusual outbound traffic'},
]
priority_order = sorted(alerts, key=lambda a: -a['severity'])
# Severity 5 first — critical alerts processed first

6. Lambda Functions & Functional Style

6.1 Lambda Syntax and Scope

# lambda args: expression
# Single expression only — no statements, no assignments, no return keyword

double   = lambda x: x * 2            # ← function object
add      = lambda x, y: x + y
identity = lambda x: x

# Default arguments
greet = lambda name, greeting="Hello": f"{greeting}, {name}!"

# Immediately Invoked Lambda (unusual but valid)
result = (lambda x, y: x ** y)(2, 10)   # 1024

# Lambdas are closures — they capture surrounding variables by reference!
multipliers = [lambda x, n=n: x * n for n in range(5)]
# Note: n=n captures the current value (default arg trick)
# Without n=n: all lambdas would capture n=4 (the final loop value)

# ❌ Common closure bug:
bad_fns = [lambda x: x * n for n in range(5)]
bad_fns[0](1)   # 4! Not 0 — n=4 (final loop value) for all of them

# ✅ Capture loop variable correctly:
good_fns = [lambda x, n=n: x * n for n in range(5)]
good_fns[0](1)  # 0
good_fns[3](1)  # 3

6.2 Lambda with Built-ins

data = [{'id': 3, 'val': 9}, {'id': 1, 'val': 25}, {'id': 2, 'val': 4}]

# sorted with lambda key
sorted(data, key=lambda d: d['val'])       # by val ascending

# max/min with lambda key
max(data, key=lambda d: d['id'])           # id=3

# map with lambda
list(map(lambda d: d['val'] ** 0.5, data)) # [3.0, 5.0, 2.0]

# filter with lambda
list(filter(lambda d: d['val'] > 5, data)) # [{'id':3,'val':9}, {'id':1,'val':25}]

# Chained functional pipeline:
result = list(
    map(
        lambda d: {**d, 'sqrt_val': round(d['val'] ** 0.5, 2)},
        filter(lambda d: d['val'] > 0, data)
    )
)

6.3 When NOT to Use Lambda

# ❌ Too complex — unreadable
process = lambda x: (x.strip().lower() if isinstance(x, str) else str(x)).replace(' ', '_')

# ✅ Use a named function — self-documenting
def normalise_field(x) -> str:
    """Convert to lowercase string with spaces as underscores."""
    if not isinstance(x, str):
        x = str(x)
    return x.strip().lower().replace(' ', '_')

# ❌ Lambda assigned to a name (PEP 8 violation — use def instead)
square = lambda x: x**2       # linters will flag this

# ✅ Named function
def square(x): return x ** 2

# Lambda is best as an inline anonymous key/transform passed to another function
# Not as a named reusable function

7. Comprehensions — list, dict, set, generator

7.1 List Comprehensions

# [expression for item in iterable if condition]

squares        = [x**2 for x in range(10)]
even_squares   = [x**2 for x in range(10) if x % 2 == 0]

# Nested comprehension — matrix operations
matrix    = [[1,2,3],[4,5,6],[7,8,9]]
flat      = [cell for row in matrix for cell in row]   # [1,2,3,4,5,6,7,8,9]
transposed = [[row[i] for row in matrix] for i in range(3)]

# Walrus operator (:=) in comprehension (Python 3.8+)
# Compute and keep only non-None results without calling the function twice
import re
lines = ["error: foo", "warning: bar", "info: baz", "error: qux"]
error_msgs = [
    m.group(1)
    for line in lines
    if (m := re.match(r'error: (.+)', line))
]
# ['foo', 'qux']

7.2 Dict and Set Comprehensions

# Dict comprehension: {key_expr: value_expr for item in iterable if condition}
word    = "mississippi"
freq    = {char: word.count(char) for char in set(word)}
# {'m':1, 'i':4, 's':4, 'p':2}

# Invert a dict (assumes unique values)
original  = {'a': 1, 'b': 2, 'c': 3}
inverted  = {v: k for k, v in original.items()}   # {1:'a', 2:'b', 3:'c'}

# Filter a dict
scores    = {'Alice': 95, 'Bob': 60, 'Carol': 82, 'Dave': 45}
passing   = {name: score for name, score in scores.items() if score >= 70}
# {'Alice': 95, 'Carol': 82}

# Normalize dict keys (security: prevent key confusion attacks)
raw_config = {'  Host ': 'localhost', 'PORT': '5432', 'Db_Name': 'mydb'}
clean_config = {k.strip().lower(): v.strip() for k, v in raw_config.items()}
# {'host': 'localhost', 'port': '5432', 'db_name': 'mydb'}

# Set comprehension
unique_domains = {email.split('@')[1].lower() for email in [
    'Alice@Example.com', 'bob@gmail.com', 'carol@EXAMPLE.COM'
]}
# {'example.com', 'gmail.com'}

7.3 Generator Expressions — Lazy Evaluation

# Generator expressions compute values ONE AT A TIME on demand
# Critical for large datasets — use O(1) memory regardless of size

# Materialise with list() or iterate directly
gen = (x**2 for x in range(10))   # NO computation yet
next(gen)   # 0 — computes first value
next(gen)   # 1 — computes second value

# Pass directly to functions that accept iterables
total = sum(x**2 for x in range(1_000_000))   # O(1) memory!
maximum = max(len(line) for line in open('file.txt'))  # line by line

# Generator with filter
safe_inputs = (
    s.strip()
    for s in raw_inputs
    if s.strip() and len(s.strip()) <= 256
)

# When to use list vs generator:
# - list comprehension: when you need the result multiple times, or need len()
# - generator expression: when you only iterate once, especially for large data

8. Exception Handling — try, except, else, finally

This is one of the most important Python topics for writing robust, secure code. Sloppy exception handling is a major source of security vulnerabilities.

8.1 The Full try/except/else/finally Anatomy

def read_config(path: str) -> dict:
    try:
        # ── Code that might raise exceptions ──
        with open(path, 'r') as f:
            content = f.read()
        config = parse_config(content)
        return config

    except FileNotFoundError:
        # ── Specific exception — handle gracefully ──
        print(f"Config file not found: {path}")
        return {}

    except PermissionError:
        # ── Different specific exception ──
        raise RuntimeError(f"Cannot read config: permission denied") from None

    except (ValueError, KeyError) as e:
        # ── Multiple exceptions in one handler ──
        raise ValueError(f"Malformed config in {path}: {e}") from e

    except Exception as e:
        # ── Catch-all for truly unexpected errors ──
        # Log it, then re-raise — don't silently swallow
        import logging
        logging.exception(f"Unexpected error reading {path}")
        raise   # re-raise the SAME exception with traceback preserved

    else:
        # ── Runs ONLY if no exception was raised in try ──
        # Perfect for code that should only run on success
        # NOT protected by the excepts above it
        log_successful_config_load(path)

    finally:
        # ── ALWAYS runs — exception or not ──
        # Use for guaranteed cleanup
        cleanup_temp_resources()

8.2 Exception Hierarchy — Know What You're Catching

# Built-in exception hierarchy (abbreviated):
# BaseException
# ├── SystemExit           ← sys.exit()
# ├── KeyboardInterrupt    ← Ctrl+C
# ├── GeneratorExit        ← generator .close()
# └── Exception            ← Almost everything else
#     ├── StopIteration    ← end of iteration
#     ├── ArithmeticError
#     │   ├── ZeroDivisionError
#     │   └── OverflowError
#     ├── LookupError
#     │   ├── IndexError   ← list[999]
#     │   └── KeyError     ← dict['missing']
#     ├── TypeError        ← wrong type for operation
#     ├── ValueError       ← right type, wrong value
#     ├── OSError (IOError, EnvironmentError)
#     │   ├── FileNotFoundError
#     │   ├── PermissionError
#     │   └── ConnectionError
#     ├── RuntimeError
#     └── AttributeError

# ❌ WRONG — catching BaseException catches SystemExit and Ctrl+C!
try:
    risky()
except BaseException:
    pass   # Ctrl+C now broken — user can't interrupt!

# ❌ WRONG — bare except has same problem
try:
    risky()
except:
    pass

# ✅ RIGHT — catch Exception if you want "everything except system signals"
try:
    risky()
except Exception as e:
    handle(e)

# ✅ BEST — catch SPECIFIC exceptions
try:
    value = int(user_input)
except ValueError:
    print("Not a valid integer")

8.3 Exception Patterns for Secure Code

# Pattern 1: Fail-safe defaults
def get_user_role(user_id: int, db) -> str:
    try:
        user = db.query(user_id)
        return user.role
    except (DatabaseError, KeyError, AttributeError):
        # On ANY error, return most restrictive role
        return 'anonymous'   # fail-safe: deny access rather than grant it

# Pattern 2: Specific exception for each failure mode
def authenticate(username: str, password: str) -> dict:
    try:
        user = find_user(username)       # may raise UserNotFoundError
        verify_password(password, user.hash)  # may raise InvalidCredentials
        if not user.active:
            raise AccountDisabledError(f"Account {username} is disabled")
        return generate_token(user)
    except (UserNotFoundError, InvalidCredentials):
        # SAME error message for both — prevents username enumeration!
        raise AuthenticationError("Invalid username or password") from None
    except AccountDisabledError:
        raise  # re-raise as-is for specific handling by caller

# Pattern 3: Never swallow exceptions silently
def bad_auth(username, password):
    try:
        return do_auth(username, password)
    except:
        return None   # ← SILENT FAILURE: bug? attack? we'll never know

def good_auth(username, password):
    try:
        return do_auth(username, password)
    except AuthError:
        raise   # propagate to caller
    except Exception as e:
        logging.error(f"Unexpected auth error: {type(e).__name__}: {e}")
        raise RuntimeError("Authentication service unavailable") from e

# Pattern 4: Context-safe exception handling
import contextlib

# Suppress a specific exception (use sparingly)
with contextlib.suppress(FileNotFoundError):
    os.remove(temp_file)   # OK if file doesn't exist — we wanted it gone anyway

8.4 The else Clause — Often Forgotten

# The else clause runs only when try completes WITHOUT raising
# This is subtly different from putting code at the end of try

# Without else — misleading: the except catches setup AND parse errors
try:
    data = fetch_data()
    result = parse(data)   # if this raises, it looks like a fetch error
except NetworkError as e:
    handle_network_error(e)

# With else — clear separation of concerns
try:
    data = fetch_data()     # only network errors caught here
except NetworkError as e:
    handle_network_error(e)
else:
    result = parse(data)    # parse errors propagate naturally (not caught)
    process(result)

9. Custom Exceptions & Exception Chaining

9.1 Designing a Custom Exception Hierarchy

# Base exception for your application — all app errors inherit from this
class AppError(Exception):
    """Base exception for all application errors."""
    def __init__(self, message: str, code: str | None = None):
        super().__init__(message)
        self.code = code
        self.message = message

    def __str__(self):
        if self.code:
            return f"[{self.code}] {self.message}"
        return self.message

# Specific error categories
class ValidationError(AppError):
    """Raised when user input fails validation."""
    pass

class AuthenticationError(AppError):
    """Raised when authentication fails."""
    def __init__(self, message: str = "Invalid credentials"):
        super().__init__(message, code="AUTH_FAILED")

class AuthorizationError(AppError):
    """Raised when authenticated user lacks permission."""
    def __init__(self, resource: str, action: str):
        super().__init__(
            f"Permission denied: cannot {action} on {resource}",
            code="AUTHZ_DENIED"
        )
        self.resource = resource
        self.action   = action

class DatabaseError(AppError):
    """Raised when database operations fail."""
    pass

class RateLimitError(AppError):
    """Raised when rate limit is exceeded."""
    def __init__(self, retry_after: int = 60):
        super().__init__(
            f"Rate limit exceeded. Try again in {retry_after}s.",
            code="RATE_LIMITED"
        )
        self.retry_after = retry_after

# Usage
def get_user(user_id: int) -> dict:
    if not isinstance(user_id, int) or user_id <= 0:
        raise ValidationError(f"Invalid user_id: {user_id!r}", code="INVALID_ID")
    user = db.find(user_id)
    if user is None:
        raise AuthenticationError()
    if not user['active']:
        raise AuthorizationError(resource=f"user/{user_id}", action="access")
    return user

9.2 Exception Chaining — raise ... from ...

# Python 3: exceptions chain automatically (implicit chaining)
try:
    int("abc")
except ValueError:
    raise RuntimeError("Config parsing failed")
# RuntimeError shows "During handling of the above exception, another exception occurred"
# The original ValueError is accessible as __context__

# Explicit chaining: raise X from Y — use when you want to document the cause
def load_config(path: str) -> dict:
    try:
        with open(path) as f:
            return json.load(f)
    except FileNotFoundError as e:
        raise AppError(f"Config not found: {path}") from e   # e is the __cause__
    except json.JSONDecodeError as e:
        raise ValidationError(f"Malformed JSON in {path}: {e}") from e

# Suppress chaining: raise X from None — hides the original
# Use when the original exception contains sensitive information
def authenticate(username: str, password: str):
    try:
        user = db.query("SELECT * FROM users WHERE username=?", (username,))
        if user is None:
            raise AuthenticationError()
    except DatabaseError as e:
        # The DatabaseError might reveal internal schema details
        # raise AuthenticationError() from e  ← LEAKS db info in traceback
        raise AuthenticationError() from None   # clean — hides db error

# Accessing chained exceptions
try:
    load_config("missing.json")
except AppError as e:
    print(e)           # [None] Config not found: missing.json
    print(e.__cause__)  # [Errno 2] No such file or directory: 'missing.json'

9.3 Exception Groups (Python 3.11+)

# ExceptionGroup lets you raise/handle multiple exceptions at once
# Useful for parallel operations that can fail independently

import asyncio

# Raising an ExceptionGroup
def validate_form(data: dict) -> None:
    errors = []
    if not data.get('email'):
        errors.append(ValueError("Email is required"))
    if not data.get('username') or len(data['username']) < 3:
        errors.append(ValueError("Username must be at least 3 characters"))
    if errors:
        raise ExceptionGroup("Form validation failed", errors)

# Handling with except* (Python 3.11+)
try:
    validate_form({'username': 'ab'})
except* ValueError as eg:
    for exc in eg.exceptions:
        print(f"Validation error: {exc}")

10. Context Managers & Resource Safety

10.1 The with Statement Contract

# The context manager protocol: __enter__ and __exit__
# __enter__: called at start, return value is bound to 'as' variable
# __exit__: ALWAYS called — even on exception, return, or Ctrl+C

# File handling — the canonical example
with open('secret.txt', 'r') as f:
    data = f.read()
# f.close() is GUARANTEED even if read() raises an exception

# Multiple context managers (Python 3.10+ — parenthesised form)
with (
    open('input.txt', 'r') as src,
    open('output.txt', 'w') as dst,
    database.transaction() as tx
):
    dst.write(src.read())
    tx.commit()

# Both are equivalent to nested withs:
with open('input.txt', 'r') as src:
    with open('output.txt', 'w') as dst:
        dst.write(src.read())

10.2 contextlib — Building Context Managers

import contextlib
import threading

# @contextmanager — simplest way to build a context manager
@contextlib.contextmanager
def managed_temp_file(suffix: str = '.tmp') -> str:
    import tempfile, os
    path = tempfile.mktemp(suffix=suffix)
    try:
        yield path          # everything BEFORE yield is __enter__
    finally:
        # everything in finally is __exit__ — ALWAYS runs
        if os.path.exists(path):
            os.remove(path)
            print(f"Cleaned up {path}")

with managed_temp_file('.json') as path:
    with open(path, 'w') as f:
        json.dump({'secret': 'data'}, f)
    process(path)
# file deleted automatically even if process() raises

# Context manager for timing
@contextlib.contextmanager
def timer(label: str):
    import time
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.4f}s")

with timer("Database query"):
    results = db.query(sql)

# Context manager for acquiring/releasing locks
@contextlib.contextmanager
def locked(lock: threading.Lock, timeout: float = 5.0):
    acquired = lock.acquire(timeout=timeout)
    if not acquired:
        raise RuntimeError("Could not acquire lock within timeout")
    try:
        yield
    finally:
        lock.release()

10.3 Security-Critical Resource Patterns

import contextlib, secrets, os

# Pattern: ensure sensitive data is cleared from memory
@contextlib.contextmanager
def sensitive_buffer():
    """Clears sensitive bytes from memory on exit."""
    data = bytearray(256)   # mutable — we can zero it out
    try:
        yield data
    finally:
        # Zero out the buffer (prevents cold-boot/memory inspection attacks)
        for i in range(len(data)):
            data[i] = 0

with sensitive_buffer() as buf:
    # read password into buf
    pass
# buf is now all zeros

# Pattern: atomic file write (prevent partial writes on crash)
@contextlib.contextmanager
def atomic_write(path: str, mode: str = 'w'):
    """Write to a temp file, then atomically rename to target."""
    tmp_path = path + f'.{secrets.token_hex(4)}.tmp'
    try:
        with open(tmp_path, mode) as f:
            yield f
        os.replace(tmp_path, path)   # atomic on POSIX systems
    except Exception:
        with contextlib.suppress(FileNotFoundError):
            os.remove(tmp_path)
        raise

with atomic_write('/etc/app/config.json') as f:
    json.dump(new_config, f, indent=2)
# Readers always see a complete file — never a partial write

11. String Security — Formatting, Sanitisation, Validation

11.1 f-strings and Format Security

# f-strings (Python 3.6+) — fast, readable, but NOT safe for untrusted templates
name   = "Alice"
score  = 95.678
status = True

# Basic f-string
print(f"Hello, {name}!")

# Format specifiers
print(f"Score: {score:.2f}")        # 95.68 — 2 decimal places
print(f"Score: {score:08.2f}")      # 00095.68 — padded
print(f"Hex: {255:#010x}")          # 0x000000ff
print(f"Big number: {1_000_000:,}") # 1,000,000

# Expressions inside f-strings
print(f"{'PASS' if status else 'FAIL'}")
print(f"{name.upper()!r}")          # 'ALICE' — !r for repr(), !s for str(), !a for ascii()

# Self-documenting (Python 3.8+) — variable name + value
x = 42
print(f"{x = }")    # x = 42 (includes variable name!)

# ❌ CRITICAL SECURITY: NEVER use f-strings with user-controlled templates
user_template = input("Enter template: ")   # user enters: {config.__class__.__dict__}
# NEVER: eval(f"{user_template}")  or  format_map with attacker-controlled keys

# ✅ Safe template system — use string.Template with $ substitution
from string import Template
t = Template("Hello, $name! Your score is $score.")
result = t.safe_substitute(name=user_name, score=95)  # safe_substitute ignores unknown keys
# User CANNOT access variables beyond what you provide

11.2 String Methods for Validation

# Know all the string methods used in security validation:

s = "Hello, World! 123"

# Character classification (per-char boolean)
"abc".isalpha()        # True — all letters
"123".isdigit()        # True — all digits (0-9)
"abc123".isalnum()     # True — letters and digits
"  ".isspace()         # True — all whitespace
"HELLO".isupper()      # True
"hello".islower()      # True
"Hello World".istitle() # True — title case

# Safer: use re for precise control
import re
re.fullmatch(r'[a-zA-Z]{2,50}', name)    # only letters, 2-50 chars
re.fullmatch(r'\d{4}-\d{2}-\d{2}', date) # YYYY-MM-DD date format
re.fullmatch(r'[^@\s]+@[^@\s]+\.[^@\s]+', email)

# Strip methods
"  hello  ".strip()     # "hello"
"  hello  ".lstrip()    # "hello  "
"  hello  ".rstrip()    # "  hello"
"xxhelloxx".strip('x')  # "hello" — strips specified chars from both ends

# Case methods
"hello".upper()         # "HELLO"
"HELLO".lower()         # "hello"
"hello world".title()   # "Hello World"
"Hello World".swapcase() # "hELLO wORLD"
"hello WORLD".casefold() # "hello world" — more aggressive than lower() for Unicode

# Membership
"hello".startswith("he")    # True
"hello".endswith("lo")      # True
"hello".startswith(("he", "wo"))  # True — tuple of prefixes

# Partitioning
"user=alice".partition('=')     # ('user', '=', 'alice')
"host:port:db".split(':', 1)    # ['host', 'port:db'] — split once only!
"host:port:db".split(':')       # ['host', 'port', 'db']

11.3 Encoding and Unicode Security

# Encoding
"hello".encode('utf-8')          # b'hello'
b"hello".decode('utf-8')         # "hello"
"café".encode('utf-8')           # b'caf\xc3\xa9'

# Always specify encoding explicitly
with open('file.txt', 'r', encoding='utf-8') as f:
    data = f.read()

# Unicode normalisation — CRITICAL for security
import unicodedata

# Homograph attack: different Unicode, same visual appearance
admin1 = "admin"       # all ASCII
admin2 = "\u0430dmin"  # Cyrillic 'а' in first position
print(admin1 == admin2)   # False! Different bytes, same look in many fonts

# Normalise before comparing
def safe_compare(a: str, b: str) -> bool:
    return unicodedata.normalize('NFKC', a) == unicodedata.normalize('NFKC', b)

# NFKC normalisation collapses many lookalike characters
print(safe_compare("admin", "ａｄｍｉｎ"))   # True — fullwidth chars normalised

# HTML escaping (prevent XSS)
import html
user_input = "<script>alert('xss')</script>"
safe       = html.escape(user_input)         # &lt;script&gt;alert(&#x27;xss&#x27;)&lt;/script&gt;
unsafe     = html.unescape(safe)             # back to original (only do this for display)

# URL encoding
from urllib.parse import quote, unquote, urlencode
safe_url = quote("hello world/path")      # "hello%20world%2Fpath"
safe_qs  = urlencode({'name': 'alice', 'q': 'hello world'})

12. Injection Attacks — SQL, Command, Path Traversal

12.1 SQL Injection — The Most Common Critical Vulnerability

import sqlite3

# ❌ VULNERABLE — string interpolation / concatenation
def get_user_bad(username: str) -> dict | None:
    conn = sqlite3.connect("app.db")
    query = f"SELECT * FROM users WHERE username = '{username}'"
    # If username = "'; DROP TABLE users; --"
    # Query becomes: SELECT * FROM users WHERE username = ''; DROP TABLE users; --'
    result = conn.execute(query).fetchone()
    return result

# ❌ Also vulnerable — .format() and % formatting
query = "SELECT * FROM users WHERE id = %s" % user_id    # WRONG
query = "SELECT * FROM users WHERE id = {}".format(user_id)  # WRONG

# ✅ SAFE — parameterized queries (placeholders)
def get_user_safe(username: str) -> dict | None:
    conn = sqlite3.connect("app.db")
    # The ? placeholder is ALWAYS safe — driver handles escaping
    cursor = conn.execute(
        "SELECT id, username, role FROM users WHERE username = ?",
        (username,)    # always a tuple, even for single param
    )
    row = cursor.fetchone()
    if row:
        return dict(zip([d[0] for d in cursor.description], row))
    return None

# ✅ Multiple parameters
def get_active_users_by_role(role: str, min_age: int) -> list:
    conn = sqlite3.connect("app.db")
    cursor = conn.execute(
        "SELECT username, email FROM users WHERE role = ? AND age >= ? AND active = 1",
        (role, min_age)
    )
    return cursor.fetchall()

# Column and table names CANNOT be parameterized — use allowlisting instead
ALLOWED_COLUMNS = {'username', 'email', 'created_at'}
ALLOWED_TABLES  = {'users', 'posts', 'comments'}

def dynamic_query(table: str, column: str) -> str:
    if table not in ALLOWED_TABLES:
        raise ValueError(f"Invalid table: {table!r}")
    if column not in ALLOWED_COLUMNS:
        raise ValueError(f"Invalid column: {column!r}")
    return f"SELECT {column} FROM {table}"   # safe because we allowlisted

12.2 Command Injection

import subprocess, shlex, os

# ❌ CRITICALLY VULNERABLE — shell=True with user input
def ping_bad(host: str) -> str:
    result = os.system(f"ping -c 1 {host}")
    # host = "google.com; cat /etc/passwd"  → executes BOTH commands!

# ❌ Also vulnerable
def ping_bad2(host: str) -> str:
    result = subprocess.run(f"ping -c 1 {host}", shell=True, capture_output=True)

# ✅ SAFE — list form, no shell
def ping_safe(host: str) -> subprocess.CompletedProcess:
    # Validate input first
    import re
    if not re.fullmatch(r'[a-zA-Z0-9.\-]{1,253}', host):
        raise ValueError(f"Invalid hostname: {host!r}")

    # List form: OS passes args directly to execve, bypassing the shell
    return subprocess.run(
        ["ping", "-c", "1", host],   # each arg is a separate list element
        capture_output=True,
        text=True,
        timeout=5              # always set timeout for external commands!
    )

# ✅ If you MUST use shell=True, use shlex.quote (POSIX systems only)
def grep_safe_but_fragile(pattern: str, filepath: str) -> str:
    # shlex.quote wraps the argument in single quotes and escapes internal quotes
    safe_pattern  = shlex.quote(pattern)
    safe_filepath = shlex.quote(filepath)
    result = subprocess.run(
        f"grep {safe_pattern} {safe_filepath}",
        shell=True, capture_output=True, text=True
    )
    return result.stdout
    # Still prefer the list form — shlex.quote has edge cases on Windows

12.3 Path Traversal

import os
from pathlib import Path

BASE_DIR = Path("/var/app/uploads").resolve()

# ❌ VULNERABLE — naive path join
def serve_file_bad(filename: str) -> bytes:
    path = os.path.join("/var/app/uploads", filename)
    # filename = "../../etc/passwd" → "/var/app/uploads/../../etc/passwd" → "/etc/passwd"
    with open(path, 'rb') as f:
        return f.read()

# ✅ SAFE — resolve and validate the final path
def serve_file_safe(filename: str) -> bytes:
    # Resolve symlinks and normalise the path
    requested = (BASE_DIR / filename).resolve()

    # The final resolved path MUST be inside BASE_DIR
    if not str(requested).startswith(str(BASE_DIR)):
        raise PermissionError(f"Access denied: {filename!r}")

    # Additional validation
    if not requested.is_file():
        raise FileNotFoundError(f"File not found: {filename!r}")

    with open(requested, 'rb') as f:
        return f.read()

# Test cases:
# serve_file_safe("report.pdf")         → OK
# serve_file_safe("../../etc/passwd")   → PermissionError
# serve_file_safe("/etc/shadow")        → PermissionError (absolute path attack)

# Using pathlib (cleaner):
def serve_file_pathlib(filename: str) -> bytes:
    safe_path = BASE_DIR / Path(filename).name  # .name strips directory components!
    # Path("../../etc/passwd").name == "passwd" — only the filename!
    if not safe_path.exists():
        raise FileNotFoundError(filename)
    return safe_path.read_bytes()

13. Password & Secret Handling

13.1 Why You Must NEVER Use SHA-256 for Passwords

import hashlib, time

# The fundamental problem: SHA-256 is DESIGNED to be fast
# Performance test (illustrative):
password = "hunter2"

start = time.perf_counter()
for _ in range(1_000_000):
    hashlib.sha256(password.encode()).hexdigest()
elapsed = time.perf_counter() - start
print(f"SHA-256: 1M hashes in {elapsed:.2f}s")
# Result: ~0.5s on a laptop CPU
# Modern GPUs: BILLIONS of SHA-256 hashes per second
# Dictionary attack on a stolen SHA-256 password hash: seconds to minutes

# Password hashing requirements:
# 1. SLOW — intentionally high work factor
# 2. SALTED — unique random value per password (prevents rainbow tables)
# 3. ADAPTIVE — work factor can be increased as hardware improves
# These properties are provided by: bcrypt, scrypt, Argon2

# ❌ WRONG password storage methods:
hashlib.md5(password.encode())          # fast + broken cryptography
hashlib.sha256(password.encode())       # fast (see above)
hashlib.sha256((password + "salt").encode())  # fast — salt doesn't fix speed

13.2 bcrypt — Industry Standard

import bcrypt   # pip install bcrypt

password = "MySecurePassword!123"

# Hash — slow by design (12 rounds = ~250ms on modern hardware)
# rounds: each increment doubles the work — 12 is a good default (2^12 iterations)
hashed = bcrypt.hashpw(
    password.encode('utf-8'),
    bcrypt.gensalt(rounds=12)
)
# b'$2b$12$...' — 60-byte string containing: algorithm, rounds, salt, hash
print(type(hashed))   # bytes
print(len(hashed))    # 60

# Verify — constant-time comparison built in
def check_password(plain: str, hashed_pw: bytes) -> bool:
    return bcrypt.checkpw(plain.encode('utf-8'), hashed_pw)

print(check_password("MySecurePassword!123", hashed))   # True
print(check_password("wrong_password", hashed))          # False

# Storage: store the full hashed bytes (or decode to str for varchar columns)
stored_hash = hashed.decode('utf-8')  # safe to store as string
retrieved   = stored_hash.encode('utf-8')  # convert back for checkpw

13.3 Python's hashlib.scrypt (Built-in)

import hashlib, os

# scrypt — built into Python 3.6+ standard library (no extra package!)
password = b"MySecurePassword!123"
salt     = os.urandom(32)   # ALWAYS use a random salt, never hardcode

# Parameters:
# n = CPU/memory cost (must be power of 2, min 2^14 for interactive)
# r = block size (8 is standard)
# p = parallelisation factor
hashed = hashlib.scrypt(password, salt=salt, n=2**14, r=8, p=1, dklen=64)
# Store salt + hashed together (both needed for verification)

def hash_password_scrypt(plain: str) -> tuple[bytes, bytes]:
    salt   = os.urandom(32)
    hashed = hashlib.scrypt(plain.encode(), salt=salt, n=2**14, r=8, p=1, dklen=64)
    return salt, hashed

def verify_password_scrypt(plain: str, salt: bytes, stored_hash: bytes) -> bool:
    computed = hashlib.scrypt(plain.encode(), salt=salt, n=2**14, r=8, p=1, dklen=64)
    return hmac.compare_digest(computed, stored_hash)   # constant-time!

13.4 Secrets Module — Cryptographically Secure Randomness

import secrets, random

# ❌ WRONG for security purposes
token_bad  = random.randint(0, 2**32)     # NOT cryptographically random!
token_bad2 = str(hash(time.time()))       # predictable, collision-prone

# ✅ CORRECT — secrets module uses OS CSPRNG (urandom)
token_hex  = secrets.token_hex(32)        # 64 hex chars = 256 bits of entropy
token_b64  = secrets.token_urlsafe(32)    # URL-safe base64 = ~43 chars
token_raw  = secrets.token_bytes(32)      # 32 raw bytes

# Use cases:
session_id   = secrets.token_urlsafe(32)  # session token
reset_token  = secrets.token_hex(32)      # password reset link token
csrf_token   = secrets.token_hex(16)      # CSRF protection
api_key      = secrets.token_urlsafe(48)  # API key

# secrets.choice — for generating passwords/pins
import string
alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
# Pick 16 characters at random — each pick uses CSPRNG
password = ''.join(secrets.choice(alphabet) for _ in range(16))

# secrets.compare_digest (wraps hmac.compare_digest) — constant-time comparison
def safe_compare_tokens(a: str, b: str) -> bool:
    return secrets.compare_digest(a, b)   # resists timing attacks

13.5 Environment Variables & Secret Management

import os
from functools import lru_cache

# ❌ NEVER hardcode secrets
DATABASE_URL = "postgresql://admin:password123@db.internal/myapp"
API_KEY      = "sk_live_abcdef123456"

# ✅ Read from environment
DATABASE_URL = os.environ["DATABASE_URL"]     # raises KeyError if missing — GOOD
API_KEY      = os.getenv("API_KEY", "")       # returns "" if missing

# ✅ Validate at startup — fail fast if config is incomplete
def load_config() -> dict:
    required = ["DATABASE_URL", "SECRET_KEY", "REDIS_URL"]
    missing  = [key for key in required if not os.getenv(key)]
    if missing:
        raise EnvironmentError(f"Missing required environment variables: {missing}")
    return {key: os.environ[key] for key in required}

# ✅ Never log secrets
import logging
secret_key = os.environ["SECRET_KEY"]
logging.info(f"Using key: {secret_key[:8]}...")   # log only prefix!
# logging.info(f"Key: {secret_key}")  ← NEVER DO THIS

# ✅ Use python-dotenv for local development only
# from dotenv import load_dotenv
# load_dotenv()   # loads .env file — .env should be in .gitignore!

14. Cryptography Fundamentals in Python

14.1 HMAC — Message Authentication Code

import hmac, hashlib, secrets

# HMAC: proves a message came from someone with the shared key AND wasn't modified
# It does NOT encrypt — the message is visible, but authenticated

key     = secrets.token_bytes(32)   # shared secret key
message = b"user_id=42;role=admin;expires=1700000000"

# Create MAC
mac = hmac.new(key, message, hashlib.sha256).digest()   # 32 bytes

# Verify on receipt — MUST use compare_digest, never ==
def verify_mac(key: bytes, message: bytes, received_mac: bytes) -> bool:
    expected = hmac.new(key, message, hashlib.sha256).digest()
    return hmac.compare_digest(expected, received_mac)  # constant-time!

# Tamper-evident cookies (simplified):
def sign_cookie(data: str, key: bytes) -> str:
    message = data.encode()
    mac     = hmac.new(key, message, hashlib.sha256).hexdigest()
    return f"{data}.{mac}"

def verify_cookie(cookie: str, key: bytes) -> str | None:
    if '.' not in cookie:
        return None
    data, received_mac = cookie.rsplit('.', 1)
    expected_mac = hmac.new(key, data.encode(), hashlib.sha256).hexdigest()
    if hmac.compare_digest(expected_mac, received_mac):
        return data
    return None   # tampered!

14.2 Symmetric Encryption — Fernet (AES-128-CBC)

from cryptography.fernet import Fernet, InvalidToken   # pip install cryptography

# Generate a key — store this securely (env var, secrets manager)
key = Fernet.generate_key()   # 32 random bytes, base64-encoded
# b'gAAAAAB...' — safe to store as a string

f = Fernet(key)

# Encrypt — returns ciphertext (bytes)
plaintext  = b"This is sensitive PII data"
ciphertext = f.encrypt(plaintext)
# Each call produces different ciphertext (random IV) — safe!

# Decrypt
try:
    decrypted = f.decrypt(ciphertext)
    print(decrypted)   # b"This is sensitive PII data"
except InvalidToken:
    print("Decryption failed — data tampered or wrong key!")

# ✅ Fernet provides:
# 1. Confidentiality — AES-128-CBC encryption
# 2. Integrity — HMAC-SHA256 authentication (detects tampering)
# 3. Freshness — includes a timestamp (optional TTL enforcement)

# With TTL (reject tokens older than 60 seconds):
try:
    decrypted = f.decrypt(ciphertext, ttl=60)
except InvalidToken:
    print("Token expired or invalid")

# Rotating keys — key rotation without losing old data
from cryptography.fernet import MultiFernet

old_key = Fernet.generate_key()
new_key = Fernet.generate_key()
mf = MultiFernet([Fernet(new_key), Fernet(old_key)])
# Encrypts with new_key, decrypts with either

14.3 Hashing for Integrity (Non-Password Use)

import hashlib

# SHA-256 IS appropriate for:
# - File integrity verification (is this download uncorrupted?)
# - Content-addressable storage
# - Generating deterministic IDs from content
# NOT appropriate for: passwords (see section 13)

def file_checksum(filepath: str) -> str:
    """Compute SHA-256 hash of a file efficiently."""
    h = hashlib.sha256()
    with open(filepath, 'rb') as f:
        # Read in chunks — handles large files without loading into memory
        for chunk in iter(lambda: f.read(65536), b''):
            h.update(chunk)
    return h.hexdigest()

# Verify a downloaded file
def verify_download(filepath: str, expected_sha256: str) -> bool:
    actual = file_checksum(filepath)
    return hmac.compare_digest(actual, expected_sha256.lower())

# Content-based cache key
def cache_key(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()[:16]  # first 16 hex chars = 64-bit key

# SHA-3 (Keccak) — alternative to SHA-2, different design
sha3_hash = hashlib.sha3_256(b"data").hexdigest()

15. Authentication — JWT, Sessions, Tokens

15.1 JWT Deep Dive

import jwt   # pip install PyJWT
import time
import os
from datetime import datetime, timezone, timedelta

SECRET_KEY = os.environ.get("JWT_SECRET", "change-me-in-production")

# JWT structure: base64url(header).base64url(payload).signature
# Header: {"alg":"HS256","typ":"JWT"}
# Payload: {"sub":"user_123","iat":...,"exp":...}
# Signature: HMAC-SHA256(header.payload, secret)
# CRITICAL: payload is NOT encrypted — only signed — anyone can decode it!

def create_access_token(user_id: str, role: str = "user") -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub":  user_id,               # subject — who this token is for
        "role": role,                  # custom claim
        "iat":  now,                   # issued at
        "exp":  now + timedelta(hours=1),   # expiration — ALWAYS set this!
        "jti":  secrets.token_hex(8),  # JWT ID — enables revocation
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def verify_access_token(token: str) -> dict:
    try:
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=["HS256"],   # ALWAYS specify allowed algorithms!
            # If you omit algorithms=, an attacker can set alg=none in the header
            # and forge any token!
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise AuthenticationError("Token has expired")
    except jwt.InvalidTokenError as e:
        raise AuthenticationError(f"Invalid token: {e}")

# Token refresh pattern
def create_refresh_token(user_id: str) -> str:
    payload = {
        "sub":  user_id,
        "type": "refresh",    # mark as refresh token
        "iat":  datetime.now(timezone.utc),
        "exp":  datetime.now(timezone.utc) + timedelta(days=30),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

15.2 JWT Vulnerabilities and Mitigations

# Vulnerability 1: Algorithm confusion (alg=none)
# An attacker strips the signature and sets alg=none in the header
# MITIGATION: always specify algorithms=["HS256"] in decode()

# Vulnerability 2: HS256 with public RSA key
# If you switch from HS256 to RS256, an attacker might send an HS256 token
# signed with the PUBLIC key (which they can see!)
# MITIGATION: be explicit about which algorithms are accepted per token type

# Vulnerability 3: No expiration
# A stolen token works forever
# MITIGATION: always set exp claim, keep it short (1h for access, 30d for refresh)

# Vulnerability 4: Sensitive data in payload
# JWT payload is just base64 — anyone with the token can read it
bad_payload = {
    "sub": "user_123",
    "password_hash": "$2b$12$...",  # ← NEVER DO THIS
    "ssn": "123-45-6789",           # ← NEVER DO THIS
    "credit_card": "4111111111111111"  # ← NEVER DO THIS
}

# Vulnerability 5: Missing issuer/audience validation
# A token from service A might be replayed against service B
payload = {
    "sub": "user_123",
    "iss": "https://auth.myapp.com",   # issuer
    "aud": "api.myapp.com",            # intended audience
}
jwt.decode(token, key, algorithms=["HS256"],
           issuer="https://auth.myapp.com",
           audience="api.myapp.com")   # validates both claims!

16. Timing Attacks & Side Channels

16.1 How Timing Attacks Work

import time, hmac, secrets

# The vulnerability: early-exit string comparison reveals information
# through execution time differences

# ❌ VULNERABLE comparison
def check_api_key_bad(submitted: str, real_key: str) -> bool:
    if len(submitted) != len(real_key):
        return False   # ← returns FAST for wrong-length input — leaks length!
    for a, b in zip(submitted, real_key):
        if a != b:
            return False   # ← returns faster when early mismatch found
    return True

# An attacker sends millions of requests with different prefixes,
# measures response time, and determines the correct key byte by byte:
#
# "aaaa..." → 0.001ms (fails at pos 0)
# "baaa..." → 0.001ms (fails at pos 0)
# "zaaa..." → 0.002ms (pos 0 matches! fails at pos 1)
# "zbaa..." → 0.002ms (fails at pos 1)
# ...
# Eventually reconstructs the full key through timing measurements

# ✅ SAFE — constant-time comparison regardless of mismatch position
def check_api_key_safe(submitted: str, real_key: str) -> bool:
    return hmac.compare_digest(submitted, real_key)

# Equivalent:
def check_api_key_safe2(submitted: str, real_key: str) -> bool:
    return secrets.compare_digest(submitted, real_key)

# How compare_digest works (simplified):
# It XORs all byte pairs and accumulates the result
# ALWAYS processes ALL bytes — no early exit
# The result is 0 only if ALL bytes match

16.2 User Enumeration — An Information Side Channel

# If your app says "user not found" vs "wrong password",
# attackers can enumerate which usernames exist!

# ❌ VULNERABLE — reveals whether username exists
def login_bad(username: str, password: str) -> dict:
    user = db.find_user(username)
    if user is None:
        return {"error": "User not found"}        # leaks: this username doesn't exist
    if not bcrypt.checkpw(password.encode(), user.password_hash):
        return {"error": "Incorrect password"}    # leaks: this username EXISTS
    return {"token": generate_token(user)}

# ✅ SAFE — same error message for both failure cases
def login_safe(username: str, password: str) -> dict:
    user = db.find_user(username)

    # ALWAYS run bcrypt.checkpw — even when user doesn't exist
    # This prevents timing differences too (bcrypt is slow)
    dummy_hash = b"$2b$12$00000000000000000000000000000000000000000000000000000000"
    if user is None:
        bcrypt.checkpw(password.encode(), dummy_hash)   # waste equal time
        raise AuthenticationError("Invalid username or password")

    if not bcrypt.checkpw(password.encode(), user.password_hash):
        raise AuthenticationError("Invalid username or password")   # SAME message!

    return {"token": generate_token(user)}

# Same principle for password reset:
# ❌ "Email not found" vs "Reset email sent"
# ✅ Always: "If that email is registered, you'll receive a reset link"

16.3 Rate Limiting and Brute Force Protection

import time
from collections import defaultdict
import threading

class RateLimiter:
    """Token bucket rate limiter — thread-safe."""

    def __init__(self, max_requests: int, window_seconds: int):
        self.max    = max_requests
        self.window = window_seconds
        self._data  = defaultdict(list)   # ip → list of timestamps
        self._lock  = threading.Lock()

    def is_allowed(self, key: str) -> tuple[bool, int]:
        """Returns (allowed, seconds_until_reset)."""
        now = time.time()
        cutoff = now - self.window

        with self._lock:
            # Remove expired timestamps
            self._data[key] = [t for t in self._data[key] if t > cutoff]

            if len(self._data[key]) >= self.max:
                oldest = self._data[key][0]
                retry_after = int(self.window - (now - oldest)) + 1
                return False, retry_after

            self._data[key].append(now)
            return True, 0

# Usage in a login endpoint:
limiter = RateLimiter(max_requests=5, window_seconds=300)  # 5 per 5 minutes

def protected_login(ip: str, username: str, password: str) -> dict:
    allowed, retry_after = limiter.is_allowed(ip)
    if not allowed:
        raise RateLimitError(retry_after=retry_after)
    return login_safe(username, password)

17. Dangerous Python Patterns to Avoid

17.1 eval, exec, and compile — Never on Untrusted Input

# eval() executes arbitrary Python expressions
# exec() executes arbitrary Python statements (including imports, class definitions)
# compile() prepares code objects for exec() or eval()

# ❌ CATASTROPHICALLY DANGEROUS
eval(input("Enter expression: "))
exec(user_provided_code_string)

# ❌ Trying to "sandbox" eval — this does NOT work
eval(expr, {"__builtins__": {}})
# Attackers bypass this with:
# ().__class__.__bases__[0].__subclasses__()  — accesses all classes via MRO
# Can still reach file system, network, etc.

# ✅ Safe alternatives for common eval use cases:
# Math expressions:
import ast
def safe_eval_math(expr: str) -> float:
    tree = ast.parse(expr, mode='eval')
    # Whitelist allowed node types
    ALLOWED = {
        ast.Expression, ast.BinOp, ast.UnaryOp, ast.Num, ast.Constant,
        ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow, ast.USub
    }
    for node in ast.walk(tree):
        if type(node) not in ALLOWED:
            raise ValueError(f"Forbidden expression: {type(node).__name__}")
    return eval(compile(tree, '<expr>', 'eval'))

safe_eval_math("2 + 2 * 3")    # 8 — OK
# safe_eval_math("__import__('os')")  → ValueError: Forbidden

# Literal data structures only:
import ast
data = ast.literal_eval("{'key': [1, 2, 3], 'flag': True}")   # safe
# Only handles: strings, bytes, numbers, tuples, lists, dicts, sets, booleans, None

17.2 Pickle — Deserialisation of Untrusted Data

import pickle, json

# ❌ CRITICAL VULNERABILITY — pickle can execute arbitrary code
class Exploit:
    def __reduce__(self):
        import os
        return (os.system, ("echo OWNED && cat /etc/passwd",))

payload = pickle.dumps(Exploit())
# pickle.loads(payload)  → executes os.system("echo OWNED...")!

# Any user-controlled data run through pickle.loads is RCE
# This includes: uploaded files, cached values from Redis, cookies, API payloads

# ✅ Use JSON for data exchange
safe_data = json.loads(user_provided_json)   # executes no code

# ✅ If you MUST use pickle, sign it first
import hmac, hashlib

def safe_pickle_dumps(obj: object, key: bytes) -> bytes:
    data = pickle.dumps(obj)
    mac  = hmac.new(key, data, hashlib.sha256).digest()
    return mac + data   # prepend MAC

def safe_pickle_loads(signed_data: bytes, key: bytes) -> object:
    mac, data = signed_data[:32], signed_data[32:]
    expected  = hmac.new(key, data, hashlib.sha256).digest()
    if not hmac.compare_digest(mac, expected):
        raise ValueError("Pickle data tampered or invalid key")
    return pickle.loads(data)   # only safe because we verified the MAC

# Alternatives to pickle: json, msgpack, protobuf, orjson

17.3 Other Dangerous Patterns

import yaml   # PyYAML

# ❌ yaml.load on untrusted data — similar to pickle!
data = yaml.load(user_input, Loader=yaml.Loader)   # DANGEROUS — can call constructors

# ✅ Use safe loader
data = yaml.safe_load(user_input)   # only basic Python types, no constructors
data = yaml.load(user_input, Loader=yaml.SafeLoader)  # explicit

# ❌ assert for security checks — disabled with -O flag!
assert user.is_authenticated, "Not authenticated"  # COMPLETELY BYPASSED with python -O

# ✅ Use proper exceptions
if not user.is_authenticated:
    raise AuthenticationError("Authentication required")

# ❌ Logging sensitive data
import logging
logging.debug(f"Processing request: user={user.username} password={password}")
logging.info(f"Auth token: {token}")

# ✅ Log what's needed, redact the rest
logging.info(f"Auth attempt for user: {user.username}")
logging.debug(f"Token prefix: {token[:8]}...")

# ❌ Wildcard imports in security-sensitive code
from crypto_utils import *   # What did you import? Can attackers control it?

# ✅ Explicit imports
from crypto_utils import encrypt, decrypt, verify_mac

# ❌ Mutable default arguments (a Python gotcha)
def add_to_blacklist(ip: str, blacklist: list = []) -> list:  # shared between calls!
    blacklist.append(ip)
    return blacklist

# ✅ Use None and create fresh default each call
def add_to_blacklist(ip: str, blacklist: list | None = None) -> list:
    if blacklist is None:
        blacklist = []
    blacklist.append(ip)
    return blacklist

18. Practice Problems with Full Solutions

Problem A — Secure User Registration Pipeline

Problem: Build a complete register_user() function that validates all inputs using built-ins (any, all, filter, map), hashes the password, and raises specific custom exceptions.

import re, bcrypt, secrets
from dataclasses import dataclass

class RegistrationError(Exception):
    def __init__(self, message: str, fields: list[str]):
        super().__init__(message)
        self.fields = fields

@dataclass
class NewUser:
    username: str
    email:    str
    password_hash: bytes

def register_user(username: str, email: str, password: str, confirm: str) -> NewUser:
    errors = []

    # Validate username
    if not re.fullmatch(r'[a-zA-Z0-9_]{3,32}', username):
        errors.append("username: 3–32 alphanumeric chars/underscores only")

    # Validate email (simplified)
    if not re.fullmatch(r'[^@\s]+@[^@\s]+\.[^@\s]+', email.strip()):
        errors.append("email: invalid format")

    # Validate password using any() and all()
    password_rules = [
        (len(password) >= 12,                          "at least 12 characters"),
        (any(c.isupper() for c in password),           "at least one uppercase letter"),
        (any(c.islower() for c in password),           "at least one lowercase letter"),
        (any(c.isdigit() for c in password),           "at least one digit"),
        (any(c in '!@#$%^&*()_+-=' for c in password),"at least one special character"),
    ]
    pw_errors = [msg for ok, msg in password_rules if not ok]
    if pw_errors:
        errors.append(f"password: requires {', '.join(pw_errors)}")

    # Confirm password matches
    if password != confirm:
        errors.append("confirm_password: passwords do not match")

    if errors:
        raise RegistrationError("Registration failed", fields=errors)

    # Hash password — only after all validation passes
    hashed = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt(rounds=12))

    return NewUser(
        username=username.strip(),
        email=email.strip().lower(),
        password_hash=hashed
    )

# Test it:
try:
    user = register_user("alice_99", "alice@example.com", "MyPass!2024ok", "MyPass!2024ok")
    print(f"Registered: {user.username}")
except RegistrationError as e:
    for field_error in e.fields:
        print(f"  ✗ {field_error}")

Problem B — Log Parser with Functional Built-ins

Problem: Parse a list of access log entries and produce a security report: top-5 IPs with most 4xx errors, sorted by count descending.

from collections import Counter

logs = [
    "192.168.1.1 - GET /login 200",
    "203.0.113.5 - POST /login 401",
    "203.0.113.5 - POST /login 401",
    "203.0.113.5 - GET /admin 403",
    "10.0.0.1 - GET /data 200",
    "203.0.113.5 - GET /etc/passwd 404",
    "192.168.1.1 - GET /api 401",
    "198.51.100.1 - POST /login 401",
    "198.51.100.1 - POST /login 401",
    "203.0.113.5 - GET /wp-admin 404",
]

def parse_entry(line: str) -> dict:
    parts = line.split()
    return {
        'ip':     parts[0],
        'method': parts[2],
        'path':   parts[3],
        'status': int(parts[4]),
    }

def security_report(logs: list[str], top_n: int = 5) -> list[dict]:
    # 1. Parse all entries
    parsed = list(map(parse_entry, logs))

    # 2. Filter to 4xx status codes only
    client_errors = list(filter(lambda e: 400 <= e['status'] < 500, parsed))

    # 3. Count errors per IP
    ip_counts = Counter(e['ip'] for e in client_errors)

    # 4. Sort by count descending, then IP for stability
    top_ips = sorted(
        ip_counts.items(),
        key=lambda pair: (-pair[1], pair[0])   # negative count → descending
    )[:top_n]

    # 5. Build report with enumerate for rank
    return [
        {'rank': rank, 'ip': ip, 'error_count': count}
        for rank, (ip, count) in enumerate(top_ips, start=1)
    ]

report = security_report(logs)
for entry in report:
    print(f"#{entry['rank']}: {entry['ip']} — {entry['error_count']} errors")
# #1: 203.0.113.5 — 4 errors
# #2: 192.168.1.1 — 1 errors
# #3: 198.51.100.1 — 2 errors  ← wait, let's sort correctly

Problem C — Exception Chain: Database Config Loader

Problem: Write a load_database_config() function that reads from an env var, parses it, and uses proper exception chaining to wrap lower-level errors in application-level errors.

import os, re
from urllib.parse import urlparse

class ConfigError(Exception):
    pass

class DatabaseConfig:
    def __init__(self, host, port, dbname, user, password):
        self.host     = host
        self.port     = port
        self.dbname   = dbname
        self.user     = user
        self.password = password

    def __repr__(self):
        return f"DatabaseConfig(host={self.host!r}, port={self.port}, db={self.dbname!r}, user={self.user!r})"

def load_database_config() -> DatabaseConfig:
    # Step 1: Get environment variable
    try:
        url_str = os.environ["DATABASE_URL"]
    except KeyError as e:
        raise ConfigError("DATABASE_URL environment variable is not set") from e

    # Step 2: Parse the URL
    try:
        url = urlparse(url_str)
        if url.scheme not in ('postgresql', 'postgres', 'mysql'):
            raise ValueError(f"Unsupported scheme: {url.scheme!r}")

        host     = url.hostname
        port     = url.port or (5432 if 'postgres' in url.scheme else 3306)
        dbname   = url.path.lstrip('/')
        user     = url.username
        password = url.password

        # Validate required components
        missing = [name for name, val in [
            ('host', host), ('dbname', dbname), ('user', user), ('password', password)
        ] if not val]

        if missing:
            raise ValueError(f"Missing components: {', '.join(missing)}")

    except ValueError as e:
        raise ConfigError(f"Invalid DATABASE_URL format: {e}") from e

    # Step 3: Validate port range
    if not (1 <= port <= 65535):
        raise ConfigError(f"Invalid port number: {port}")

    return DatabaseConfig(host=host, port=port, dbname=dbname, user=user, password=password)

# Test:
os.environ["DATABASE_URL"] = "postgresql://admin:secret@db.local:5432/myapp"
try:
    cfg = load_database_config()
    print(cfg)
except ConfigError as e:
    print(f"Config error: {e}")
    if e.__cause__:
        print(f"  Caused by: {e.__cause__}")

Problem D — Custom Context Manager: Audit Logging

Problem: Write a context manager audit_action() that logs the start, end, duration, and success/failure of any operation, writing to a tamper-evident audit log.

import contextlib, time, hmac, hashlib, json, os
from datetime import datetime, timezone

AUDIT_KEY = secrets.token_bytes(32)   # in production: load from secure store

def sign_record(record: dict, key: bytes) -> str:
    """Create a signed audit record."""
    payload = json.dumps(record, sort_keys=True, default=str).encode()
    mac = hmac.new(key, payload, hashlib.sha256).hexdigest()
    return json.dumps({**record, '_mac': mac}, default=str)

@contextlib.contextmanager
def audit_action(action: str, user_id: str, resource: str, log_file: str = "audit.log"):
    record = {
        'timestamp': datetime.now(timezone.utc).isoformat(),
        'action':    action,
        'user_id':   user_id,
        'resource':  resource,
        'status':    'started',
    }
    start = time.perf_counter()

    try:
        yield   # run the protected code

        record['status']   = 'success'
        record['duration'] = round(time.perf_counter() - start, 4)

    except Exception as e:
        record['status']      = 'failure'
        record['error_type']  = type(e).__name__
        record['error_msg']   = str(e)
        record['duration']    = round(time.perf_counter() - start, 4)
        raise   # re-raise — don't swallow

    finally:
        # Always write the audit log entry
        signed = sign_record(record, AUDIT_KEY)
        with open(log_file, 'a') as f:
            f.write(signed + '\n')

# Usage:
with audit_action("file_download", user_id="u123", resource="/reports/q4.pdf"):
    # simulate file serving
    time.sleep(0.01)
    print("File served")

try:
    with audit_action("admin_panel", user_id="u456", resource="/admin"):
        raise PermissionError("Insufficient privileges")
except PermissionError as e:
    print(f"Blocked: {e}")

Problem E — Functional Pipeline: Security Event Processor

Problem: Using zip, enumerate, sorted, filter, map, any, all, and sum, build a pipeline that scores and prioritises security events.

events = [
    {'type': 'login_fail',    'ip': '203.0.113.5', 'user': 'root',  'count': 50},
    {'type': 'login_ok',      'ip': '10.0.0.1',    'user': 'alice', 'count': 1},
    {'type': 'port_scan',     'ip': '203.0.113.5', 'user': None,    'count': 1000},
    {'type': 'data_exfil',    'ip': '192.168.1.50','user': 'bob',   'count': 3},
    {'type': 'login_fail',    'ip': '198.51.100.1','user': 'admin', 'count': 12},
    {'type': 'config_change', 'ip': '10.0.0.5',    'user': 'carol', 'count': 1},
]

SEVERITY = {
    'login_fail':    2,
    'port_scan':     3,
    'data_exfil':    5,
    'config_change': 3,
    'login_ok':      0,
}
HIGH_VALUE_USERS = {'root', 'admin', 'carol'}

def score_event(e: dict) -> int:
    base     = SEVERITY.get(e['type'], 1)
    count_factor = min(e['count'] // 10, 10)   # cap at 10
    user_bonus   = 2 if e.get('user') in HIGH_VALUE_USERS else 0
    return base + count_factor + user_bonus

# 1. Filter out benign events
suspicious = list(filter(lambda e: SEVERITY.get(e['type'], 1) > 0, events))

# 2. Score and annotate each event using map
scored = list(map(lambda e: {**e, 'score': score_event(e)}, suspicious))

# 3. Sort by score descending
prioritised = sorted(scored, key=lambda e: -e['score'])

# 4. Enumerate with rank for the report
report = [(rank, e) for rank, e in enumerate(prioritised, start=1)]

# 5. Summary stats using sum and any/all
total_score  = sum(e['score'] for e in scored)
has_critical = any(e['score'] >= 10 for e in scored)
all_internal = all(e['ip'].startswith('10.') for e in scored)

print("=== SECURITY EVENT REPORT ===")
for rank, event in report:
    print(f"  #{rank} [{event['score']:2d}] {event['type']:<15} ip={event['ip']:<15} user={event.get('user','N/A')}")

print(f"\nTotal risk score: {total_score}")
print(f"Critical events present: {has_critical}")
print(f"All internal: {all_internal}")

# 6. Group by IP using zip and dict
ips     = [e['ip'] for e in scored]
scores  = [e['score'] for e in scored]
from collections import defaultdict
ip_risk = defaultdict(int)
for ip, score in zip(ips, scores):
    ip_risk[ip] += score

worst_ip = max(ip_risk, key=ip_risk.get)
print(f"\nHighest risk IP: {worst_ip} (total score {ip_risk[worst_ip]})")

19. Assessment Cheat Sheet

Built-in Functions — Signatures and Returns

Function	Signature	Returns	Notes
`zip`	`zip(*iterables)`	lazy iterator of tuples	stops at shortest
`enumerate`	`enumerate(it, start=0)`	lazy iterator of (int, item)	start defaults to 0
`map`	`map(fn, *iterables)`	lazy iterator	multiple iterables = zip-like
`filter`	`filter(fn, iterable)`	lazy iterator	`filter(None, it)` removes falsy
`sum`	`sum(it, start=0)`	number	`start=[]` flattens (slow!)
`any`	`any(iterable)`	bool	short-circuits on first True
`all`	`all(iterable)`	bool	short-circuits on first False
`min`	`min(it, key=None, default=...)`	element	raises on empty without default
`max`	`max(it, key=None, default=...)`	element	key transforms before comparing
`sorted`	`sorted(it, key=None, reverse=False)`	new list	stable, works on any iterable
`reversed`	`reversed(seq)`	lazy iterator	requires sequence (has len+index)

Exception Handling Rules

try:
    ...          # might raise
except Specific as e:
    ...          # handle one type
except (A, B) as e:
    ...          # handle multiple
except Exception as e:
    ...          # catch-all (NOT BaseException!)
    raise        # re-raise without losing traceback
else:
    ...          # runs ONLY if try succeeded — not protected by excepts
finally:
    ...          # ALWAYS runs — cleanup goes here

Exception Chaining

Syntax	Effect	Use when
`raise X from Y`	Sets `X.__cause__ = Y`	Y is the documented root cause
`raise X from None`	Suppresses chain	Original reveals sensitive info
`raise X` (inside except)	Sets `X.__context__ = Y`	Automatic implicit chaining
`raise` (bare)	Re-raises current	Want to log but propagate

Security Quick Reference

Risk	Vulnerable Code	Safe Code
SQL Injection	`f"WHERE name='{name}'"`	`cursor.execute("WHERE name=?", (name,))`
Command Injection	`os.system(f"ls {path}")`	`subprocess.run(["ls", path])`
Path Traversal	`open(base + user_path)`	Resolve + verify prefix
Password Storage	`sha256(password)`	`bcrypt.hashpw(pw, gensalt(12))`
Random Tokens	`random.randint(...)`	`secrets.token_hex(32)`
Code Execution	`eval(user_input)`	`ast.literal_eval(...)`
Pickle RCE	`pickle.loads(untrusted)`	`json.loads(untrusted)`
Token Comparison	`a == b`	`hmac.compare_digest(a, b)`
User Enumeration	Different error per case	Same error always
Secret Storage	Hardcoded in source	`os.environ["SECRET"]`

Python 3.13 Specifics

Feature	Details
`input()`	Always returns `str`, never evaluates — reinforced in 3.13
New REPL	Multi-line edit, F1=help, F2=clean copy, F3=paste mode
Free-threaded	Experimental `--disable-gil` build; use explicit locks
Error messages	`NameError`/`AttributeError` now suggest corrections
`type X = ...`	Type alias statement (stabilised from 3.12)
`def fn[T](...)`	Inline type parameter syntax

Lambda Patterns

# ✅ Use lambda for: short, inline, anonymous key/transform functions
sorted(data, key=lambda d: (d['priority'], d['name']))
filter(lambda x: x > 0, data)
max(users, key=lambda u: u.score)

# ❌ Avoid lambda for: complex logic, named reusable functions, multi-line
square = lambda x: x**2   # use def instead

# Closure gotcha — capture loop variable
# ❌ Bug:    [lambda x: x*n for n in range(5)]
# ✅ Fixed:  [lambda x, n=n: x*n for n in range(5)]

Common Mistakes in Assessments

any([]) returns False; all([]) returns True — know these edge cases
zip stops at shortest — use zip_longest if you want all elements
map/filter return iterators — wrap in list() if you need to index or iterate multiple times
sorted() returns a list; .sort() returns None — don't do x = lst.sort()
lambda in a loop captures the variable, not its value — use default argument n=n
Bare except: catches SystemExit and Ctrl+C — always use except Exception:
else in try/except runs only when NO exception occurred — not the same as code after try block
raise from None suppresses exception chain — know when to use it
input() always returns str — explicit conversion required: int(input("N: "))
hmac.compare_digest not == for all security-sensitive comparisons

PyTorch Tensors & Matrix Operations: A Deep-Dive Tutorial

Ajit Kumar — Fri, 13 Mar 2026 04:45:42 +0000

Assessment Focus: Python 3 · PyTorch · Matrix manipulations (not model training)
Level: Beginner → Advanced — read end-to-end before your test
Time to complete: ~90 minutes reading + practice

What is PyTorch and Why Does It Matter?
Tensor Fundamentals — Creation & Inspection
Data Types, Devices & Casting
Indexing, Slicing & Boolean Masking
Reshaping, Views & Squeezing
Element-wise Operations & Reductions
Broadcasting — The Silent Multiplier
Matrix Multiplication — The Core Operation
Batched Matrix Operations
Einstein Summation (einsum)
Linear Algebra — linalg Module
Advanced Indexing — gather & scatter
Memory Layout, Strides & Contiguity
Practice Problems with Full Solutions
Assessment Cheat Sheet

1. What is PyTorch and Why Does It Matter?

PyTorch is a numerical computation library built on top of C++/CUDA, exposing a Python API that feels like NumPy but with two killer features:

GPU acceleration — tensors can live on CUDA devices and perform operations at GPU speed
Autograd — automatic differentiation through any computation graph (used for training, but not the focus of matrix-only assessments)

For this assessment, think of PyTorch as NumPy with superpowers: the same mental model of n-dimensional arrays, but with broadcasting, batched operations, and linear algebra built in at native speed.

import torch
import torch.nn.functional as F

# Confirm version
print(torch.__version__)   # e.g. 2.2.0
print(torch.cuda.is_available())   # True if GPU available

Key mindset: Every operation in PyTorch either returns a new tensor (functional) or modifies in-place (trailing underscore, e.g. add_). Always know which one you're calling.

2. Tensor Fundamentals — Creation & Inspection

2.1 Ways to Create Tensors

Understanding how a tensor is created determines whether it shares memory with Python/NumPy data or owns its own memory.

import torch
import numpy as np

# ── From Python data ────────────────────────────────────────────
a = torch.tensor([1.0, 2.0, 3.0])        # copies data → always safe
b = torch.tensor([[1, 2], [3, 4]])         # 2D from nested list

# ── Sharing memory with NumPy ───────────────────────────────────
arr = np.array([1.0, 2.0, 3.0])
t   = torch.from_numpy(arr)       # SHARES memory! mutation propagates both ways
t2  = torch.as_tensor(arr)        # same — shares when possible
t3  = torch.tensor(arr)           # COPIES — fully independent

arr[0] = 99.0
print(t[0])   # tensor(99.) — changed!
print(t3[0])  # tensor(1.)  — unchanged

# ── Structured tensors ──────────────────────────────────────────
zeros    = torch.zeros(3, 4)               # all 0.0, shape (3,4)
ones     = torch.ones(2, 3, 4)            # all 1.0, shape (2,3,4)
eye      = torch.eye(4)                   # 4×4 identity matrix
empty    = torch.empty(3, 3)              # uninitialized (garbage values!)
full     = torch.full((2, 3), fill_value=7.0)  # fill with 7.0

# ── Sequence tensors ────────────────────────────────────────────
arange   = torch.arange(0, 10, step=2)   # [0, 2, 4, 6, 8]
linspace = torch.linspace(0, 1, steps=5) # [0.0, 0.25, 0.5, 0.75, 1.0]

# ── Random tensors ──────────────────────────────────────────────
torch.manual_seed(42)                   # reproducibility
uniform  = torch.rand(3, 3)             # Uniform[0, 1)
normal   = torch.randn(3, 3)            # Normal(0, 1)
randint  = torch.randint(0, 10, (3, 3)) # integers in [0, 10)
perm     = torch.randperm(8)            # random permutation of [0..7]

# ── Tensors like another tensor (same shape/dtype/device) ───────
x = torch.randn(2, 3)
torch.zeros_like(x)
torch.ones_like(x)
torch.rand_like(x)
torch.full_like(x, fill_value=5.0)

2.2 Inspecting a Tensor

Every tensor carries metadata you should query habitually:

t = torch.randn(3, 4, 5)

print(t.shape)       # torch.Size([3, 4, 5]) — like a tuple
print(t.size())      # same as .shape
print(t.size(0))     # 3 — size of dimension 0
print(t.ndim)        # 3 — number of dimensions
print(t.dtype)       # torch.float32
print(t.device)      # device(type='cpu') or device(type='cuda', index=0)
print(t.numel())     # 60 — total element count (3 * 4 * 5)
print(t.is_contiguous())  # True (freshly created tensors are contiguous)
print(t.stride())    # (20, 5, 1) — elements to skip to reach next in each dim

✏️ Quick Check 2.2

Q: What does torch.empty(3, 3) contain?

A: Uninitialized memory — whatever bytes happen to be at those addresses. It's the fastest creation method (no fill), but never use its values without first writing to the tensor. Use torch.zeros() when you need a defined starting state.

3. Data Types, Devices & Casting

3.1 The dtype Hierarchy

# Floating point (most ML operations use these)
torch.float16  # fp16 — half precision (2 bytes)
torch.float32  # fp32 — DEFAULT for tensors (4 bytes)
torch.float64  # fp64 — double precision (8 bytes)
torch.bfloat16 # brain float — better range than fp16, same size

# Integer
torch.int8     # (1 byte, signed)
torch.int16    # (2 bytes)
torch.int32    # (4 bytes)
torch.int64    # (8 bytes) — DEFAULT for integer tensors

# Boolean
torch.bool     # True/False (1 byte)

# Explicit dtype at creation
t = torch.ones(3, dtype=torch.float64)
i = torch.tensor([1, 2, 3], dtype=torch.int32)

# Casting (creates a new tensor with converted values)
t_fp32  = t.float()           # → float32
t_fp16  = t.half()            # → float16
t_fp64  = t.double()          # → float64
t_int   = t.long()            # → int64
t_bool  = t.bool()            # → bool

# Generic cast
t.to(torch.float16)
t.to(dtype=torch.int32)

3.2 Devices

cpu_tensor = torch.randn(3, 3)                           # on CPU (default)
gpu_tensor = torch.randn(3, 3, device='cuda')            # on GPU
gpu_tensor = torch.randn(3, 3, device='cuda:0')          # explicit GPU index

# Moving between devices
on_gpu = cpu_tensor.to('cuda')        # CPU → GPU
on_cpu = gpu_tensor.to('cpu')         # GPU → CPU
on_gpu = cpu_tensor.cuda()            # shorthand
on_cpu = gpu_tensor.cpu()             # shorthand

# Best practice: use device variable so code works on CPU or GPU
device = 'cuda' if torch.cuda.is_available() else 'cpu'
t = torch.randn(3, 3, device=device)

Assessment rule: All tensors in an operation must be on the same device and use compatible dtypes. Mixing CPU + GPU, or int + float in some operations, raises errors.

3.3 Mixed Operations & Promotion

a = torch.tensor([1.0])    # float32
b = torch.tensor([2.0], dtype=torch.float64)

# PyTorch promotes to the higher-precision type
c = a + b   # float64 — float32 was promoted

# Integer + float → float
d = torch.tensor([1]) + torch.tensor([1.5])  # float32

✏️ Quick Check 3

Q: You have a = torch.randn(3) (float32) and b = torch.randint(0, 5, (3,)) (int64). What happens when you compute a + b?

A: PyTorch promotes int64 to float32, so the result is float32. The rule is: if one operand is floating point, the result is floating point.

4. Indexing, Slicing & Boolean Masking

This is one of the most tested areas because the distinction between views (shared memory) and copies (independent memory) is subtle and important.

4.1 Basic Indexing and Slicing

t = torch.arange(24).reshape(2, 3, 4)
# Shape: (2, 3, 4) — think: 2 matrices, each 3 rows × 4 cols
# t[batch, row, col]

# ── Single-element indexing ─────────────────────────────────────
t[0]          # first matrix → shape (3, 4)
t[-1]         # last along dim 0 → shape (3, 4)
t[0, 1]       # row 1 of first matrix → shape (4,)
t[0, 1, 2]    # scalar element → tensor(6)
t[0][1][2]    # equivalent but slower (three indexing ops)

# ── Slicing (Python-style, half-open intervals) ─────────────────
t[:, 0, :]    # all batches, row 0, all cols → shape (2, 4)
t[..., -1]    # last column across all dims → shape (2, 3)
t[:, 1:3, :]  # rows 1 and 2 from both batches → shape (2, 2, 4)
t[:, ::2, :]  # every other row → shape (2, 2, 4)

# The Ellipsis (...) fills in as many : as needed
# t[..., 2] ≡ t[:, :, 2] for a 3D tensor

4.2 Views vs Copies — Critical Distinction

t = torch.arange(12, dtype=torch.float32).reshape(3, 4)
# [[0,1,2,3],[4,5,6,7],[8,9,10,11]]

# ── SLICES return VIEWS (shared memory) ────────────────────────
s = t[0, :]     # first row — this is a VIEW
s[0] = 999.0    # modifies t as well!
print(t[0, 0])  # tensor(999.) — YES, t changed

# Verify they share storage
print(t.storage().data_ptr() == s.storage().data_ptr())  # True

# ── Get an independent copy ─────────────────────────────────────
c = t[0, :].clone()    # COPY — fully independent
c[0] = -1.0
print(t[0, 0])  # tensor(999.) — t unchanged

# ── .detach() vs .clone() ──────────────────────────────────────
# .detach() — shares memory, removes from autograd graph
# .clone()  — copies memory, preserves grad_fn
# .clone().detach() — copy + remove from graph (common pattern)

4.3 Fancy Indexing (Always Returns a Copy)

t = torch.arange(20).reshape(4, 5)

# Index with a list of indices — picks specific rows
rows = t[[0, 2, 3]]           # rows 0, 2, 3 → shape (3, 5)

# Index with parallel lists for each dimension
rows_idx = torch.tensor([0, 1, 3])
cols_idx = torch.tensor([2, 4, 1])
t[rows_idx, cols_idx]   # picks (0,2), (1,4), (3,1) — shape (3,)
                         # t[0,2]=2, t[1,4]=9, t[3,1]=16

# These ALWAYS produce a copy — safe to mutate

4.4 Boolean Masking

t = torch.tensor([[1., -2., 3.], [-4., 5., -6.]])

# Create a boolean mask
mask = t > 0
# tensor([[True, False, True], [False, True, False]])

# Apply mask — returns 1D tensor of selected values
positives = t[mask]     # tensor([1., 3., 5.])

# Count True values
mask.sum()              # tensor(3)
mask.float().mean()     # tensor(0.5) — 50% positive

# Conditional replacement (in-place)
t[t < 0] = 0.0         # ReLU!
print(t)
# tensor([[1., 0., 3.], [0., 5., 0.]])

# torch.where — vectorised ternary (NOT in-place, returns new tensor)
result = torch.where(mask, t, torch.zeros_like(t))

✏️ Practice Problem 4

Problem: Given t = torch.arange(1, 26).reshape(5, 5).float(), write a single expression that:

Selects the bottom-right 3×3 submatrix
Sets all values in that submatrix above 20 to 0

Solution:

t = torch.arange(1, 26).reshape(5, 5).float()
sub = t[2:, 2:]      # bottom-right 3×3: [[13,14,15],[18,19,20],[23,24,25]]

# sub is a VIEW — mutations propagate to t!
sub[sub > 20] = 0.0

# t is now:
# [[ 1,  2,  3,  4,  5],
#  [ 6,  7,  8,  9, 10],
#  [11, 12, 13, 14, 15],
#  [16, 17, 18, 19, 20],
#  [21, 22,  0,  0,  0]]

Key insight: The slice t[2:, 2:] is a view, so sub[sub > 20] = 0.0 modifies t directly. This is intentional and efficient — no copy needed.

5. Reshaping, Views & Squeezing

Shape manipulation is the connective tissue of tensor math — you'll do it constantly to align tensors for operations.

5.1 reshape vs view

t = torch.arange(12, dtype=torch.float32)   # shape (12,)

# reshape — safe, handles non-contiguous, may or may not share memory
r1 = t.reshape(3, 4)      # (3, 4)
r2 = t.reshape(2, 2, 3)   # (2, 2, 3)
r3 = t.reshape(4, -1)     # (4, 3) — -1 infers the size

# view — faster but requires contiguous memory; always shares memory
v = t.view(3, 4)          # fine — t is contiguous
t_T = t.reshape(3, 4).T   # transpose is non-contiguous
# t_T.view(12)             # RuntimeError! non-contiguous
t_T.contiguous().view(12) # fix: make contiguous first, then view

# When in doubt: use reshape (safe), not view

5.2 Adding and Removing Dimensions

t = torch.tensor([1., 2., 3.])   # shape (3,)

# unsqueeze — add a dimension of size 1
t.unsqueeze(0)    # (1, 3) — new batch dimension
t.unsqueeze(1)    # (3, 1) — new column dimension
t.unsqueeze(-1)   # (3, 1) — same as unsqueeze(1) for 1D

# squeeze — remove dimensions of size 1
x = torch.randn(1, 3, 1, 4)
x.squeeze()         # (3, 4) — removes ALL size-1 dims
x.squeeze(0)        # (3, 1, 4) — removes only dim 0
x.squeeze(2)        # (1, 3, 4) — removes only dim 2
x.squeeze(-1)       # (1, 3, 1) — no wait: that removes dim 3 → (1, 3, 1)?
# Careful: squeeze(dim) only removes that dim IF its size is 1; no-op otherwise

# None indexing — alternative unsqueeze
t[None, :]    # (1, 3) — same as t.unsqueeze(0)
t[:, None]    # (3, 1) — same as t.unsqueeze(1)

5.3 Concatenation and Stacking

a = torch.randn(3, 4)
b = torch.randn(3, 4)

# cat — join along an EXISTING dimension
torch.cat([a, b], dim=0)    # (6, 4) — stack vertically
torch.cat([a, b], dim=1)    # (3, 8) — stack horizontally

# stack — create a NEW dimension
torch.stack([a, b], dim=0)  # (2, 3, 4) — batch of 2 matrices
torch.stack([a, b], dim=1)  # (3, 2, 4)
torch.stack([a, b], dim=2)  # (3, 4, 2)

# vstack/hstack — convenience wrappers
torch.vstack([a, b])   # (6, 4) — same as cat(dim=0)
torch.hstack([a, b])   # (3, 8) — same as cat(dim=1)

# split — inverse of cat
parts = torch.split(torch.cat([a, b], dim=0), split_size_or_sections=3, dim=0)
# (tensor(3,4), tensor(3,4)) — splits evenly

# chunk — split into N equal pieces
chunks = torch.chunk(torch.cat([a, b], dim=0), chunks=2, dim=0)

5.4 Transpose and Permute

t = torch.randn(2, 3, 4)

# transpose — swap exactly two dimensions
t.transpose(0, 1)   # (3, 2, 4)
t.transpose(1, 2)   # (2, 4, 3)
t.T                 # full reversal of dims — only well-defined for 2D!

# permute — reorder ALL dimensions at once
t.permute(2, 0, 1)  # (4, 2, 3) — dim 2 goes to position 0, etc.
t.permute(0, 2, 1)  # (2, 4, 3) — swap dims 1 and 2

# IMPORTANT: transpose and permute return non-contiguous tensors
print(t.transpose(0,1).is_contiguous())  # False
# Fix with .contiguous() when needed:
t.permute(2, 0, 1).contiguous()

✏️ Practice Problem 5

Problem: You receive a batch of images in NHWC format (batch, height, width, channels) with shape (8, 32, 32, 3). PyTorch convolutions expect NCHW format (batch, channels, height, width). Convert it.

Solution:

images_nhwc = torch.randn(8, 32, 32, 3)
# Axes:                     N   H   W   C
# Desired:                  N   C   H   W
# So we move axis 3 (C) to position 1

images_nchw = images_nhwc.permute(0, 3, 1, 2)
print(images_nchw.shape)   # torch.Size([8, 3, 32, 32])

# Or equivalently:
images_nchw = images_nhwc.transpose(1, 3).transpose(2, 3)

# After permute, you likely need contiguous for convolution:
images_nchw = images_nchw.contiguous()

6. Element-wise Operations & Reductions

6.1 Arithmetic Operations

All arithmetic operators work element-wise and require tensors of compatible shapes (or broadcastable shapes — covered in section 7).

a = torch.tensor([1., 2., 3., 4.])
b = torch.tensor([2., 2., 2., 2.])

# Standard arithmetic — all return NEW tensors
a + b          # [3., 4., 5., 6.]
a - b          # [-1., 0., 1., 2.]
a * b          # [2., 4., 6., 8.]  — ELEMENT-WISE multiply (not dot!)
a / b          # [0.5, 1., 1.5, 2.]
a ** 2         # [1., 4., 9., 16.] — element-wise power
a // b         # [0., 1., 1., 2.]  — floor division
a % b          # [1., 0., 1., 0.]  — modulo

# In-place variants (trailing underscore — modifies a!)
a.add_(b)      # a += b
a.mul_(2)      # a *= 2
a.sub_(1)      # a -= 1

# Functional API (equivalent, often needed for clarity)
torch.add(a, b)
torch.mul(a, b)
torch.div(a, b)
torch.pow(a, 2)

# Scalar operations — broadcast the scalar to all elements
a + 10
a * 0.5
a ** 0.5       # element-wise square root via **, equivalent to torch.sqrt(a)

6.2 Math Functions

t = torch.tensor([-2., -1., 0., 1., 2.])

torch.abs(t)             # [2., 1., 0., 1., 2.]
torch.sqrt(torch.abs(t)) # requires non-negative input
torch.exp(t)             # e^t element-wise
torch.log(t + 3)         # ln — watch out: log(0) = -inf, log(-x) = nan
torch.log2(t + 3)
torch.log10(t + 3)
torch.sin(t); torch.cos(t); torch.tanh(t)
torch.sigmoid(t)         # 1/(1+e^(-t)) — same as F.sigmoid
torch.relu(t)            # max(0, t) — equivalent to t.clamp(min=0)
torch.clamp(t, min=-1, max=1)  # clip values to [-1, 1]
torch.floor(t); torch.ceil(t); torch.round(t)
torch.sign(t)            # -1., -1., 0., 1., 1.
torch.reciprocal(t + 3)  # 1 / (t+3) element-wise

# Useful for numerical stability
torch.clamp(x, min=1e-8)  # prevent log(0)

6.3 Reduction Operations

Reductions collapse dimensions. Understanding the dim parameter is crucial.

t = torch.tensor([[1., 2., 3.],
                   [4., 5., 6.]])
# Shape: (2, 3)

# Global reductions — collapse everything to a scalar
t.sum()           # tensor(21.)
t.mean()          # tensor(3.5)
t.max()           # tensor(6.) — the value
t.min()           # tensor(1.)
t.std()           # tensor(1.8708...)
t.var()           # tensor(3.5)
t.prod()          # tensor(720.)  — product of all elements
t.norm()          # Frobenius norm √(sum of squares)

# Reduce along a dimension — 'dim' specifies which axis to collapse
t.sum(dim=0)      # tensor([5., 7., 9.]) — shape (3,): sum each column
t.sum(dim=1)      # tensor([6., 15.])   — shape (2,): sum each row
t.sum(dim=1, keepdim=True)  # tensor([[6.],[15.]]) — shape (2,1): keeps dims!

# argmax / argmin — return INDEX of max/min, not the value
t.argmax()        # tensor(5) — flat index (row-major)
t.argmax(dim=0)   # tensor([1, 1, 1]) — row index of max in each column
t.argmax(dim=1)   # tensor([2, 2])   — col index of max in each row

# Returns both value and index
vals, idxs = t.max(dim=1)   # vals=[3.,6.], idxs=[2, 2]
vals, idxs = t.min(dim=0)   # vals=[1.,2.,3.], idxs=[0,0,0]

# Cumulative reductions
t.cumsum(dim=1)   # cumulative sum along rows
t.cumprod(dim=0)  # cumulative product along columns

# Any / All — boolean reductions
(t > 3).any()     # tensor(True)  — any element > 3?
(t > 0).all()     # tensor(True)  — all elements > 0?
(t > 3).any(dim=1)  # tensor([False, True]) — per row

✏️ Practice Problem 6

Problem: Given a matrix scores of shape (100, 10) representing 100 students' scores on 10 tests, compute:

Each student's average score
Each test's average score
Number of students who passed (average ≥ 60)
Normalise the scores so each student has zero mean and unit std

Solution:

torch.manual_seed(0)
scores = torch.randint(30, 100, (100, 10)).float()

# 1. Each student's average — collapse the test dimension
student_avg = scores.mean(dim=1)    # shape (100,)

# 2. Each test's average — collapse the student dimension
test_avg    = scores.mean(dim=0)    # shape (10,)

# 3. Count passing students
passed = (student_avg >= 60).sum()  # scalar tensor
print(f"Passed: {passed.item()}")

# 4. Normalise each student's scores independently
# Need keepdim=True so broadcasting works across the 10-test dimension
mean = scores.mean(dim=1, keepdim=True)   # (100, 1)
std  = scores.std(dim=1, keepdim=True)    # (100, 1)
std  = torch.clamp(std, min=1e-8)          # avoid division by zero
normalised = (scores - mean) / std         # (100, 10) — broadcast!

# Verify: each student now has approx mean=0, std=1
print(normalised.mean(dim=1)[:5])  # near zero
print(normalised.std(dim=1)[:5])   # near one

7. Broadcasting — The Silent Multiplier

Broadcasting is PyTorch's mechanism to perform operations between tensors of different shapes without copying data. Mastering it eliminates a huge class of shape errors.

7.1 The Broadcasting Rules

Two shapes are broadcast-compatible if, aligning from the right (trailing dimensions), every pair of sizes is either:

Equal, or
One of them is 1 (the size-1 dimension expands to match the other)

If one tensor has fewer dimensions, it's treated as if it has leading size-1 dimensions prepended.

Shape (4, 3)  +  Shape (3,)
Right-align:    (4, 3)
                (   3)   ← padded to (1, 3) implicitly
               ─────────
Dim 0:  4 vs 1 → expand (1,3) row to fill 4 rows
Dim 1:  3 vs 3 → equal
Result: (4, 3)  ✓

Shape (4, 1)  +  Shape (1, 3)
Dim 0: 4 vs 1 → expand to 4
Dim 1: 1 vs 3 → expand to 3
Result: (4, 3)  ✓

Shape (2, 3)  +  Shape (4, 3)
Dim 0: 2 vs 4 → NEITHER is 1 → ERROR ✗

7.2 Practical Broadcasting Examples

# ── Example 1: add a bias to every row ─────────────────────────
weight = torch.randn(4, 5)    # 4 neurons, each with 5 inputs
bias   = torch.randn(5)       # one bias per input (not realistic but illustrative)
# Actually: bias per OUTPUT neuron
bias   = torch.randn(4)       # 4 biases, one per output
output = weight.sum(dim=1) + bias   # (4,) + (4,) — trivial, same shape

# Real neural layer: (batch, in_features) @ (in_features, out_features) + (out_features,)
X = torch.randn(32, 10)         # 32 samples, 10 features
W = torch.randn(10, 4)          # weights
b = torch.randn(4)              # one bias per output neuron
out = X @ W + b    # (32,4) + (4,) — broadcast adds b to every row of X@W

# ── Example 2: column-wise normalisation ───────────────────────
data = torch.randn(100, 5)       # 100 samples, 5 features
col_mean = data.mean(dim=0)      # (5,)     — mean of each feature
col_std  = data.std(dim=0)       # (5,)
normalised = (data - col_mean) / col_std   # (100,5) - (5,) → broadcast!

# ── Example 3: pairwise distances ──────────────────────────────
# Points: A shape (N, D), B shape (M, D)
# Distance matrix: shape (N, M)
A = torch.randn(10, 3)   # 10 points in 3D
B = torch.randn(7, 3)    # 7 points in 3D

# A.unsqueeze(1): (10, 1, 3)
# B.unsqueeze(0): (1, 7, 3)
# Subtraction:    (10, 7, 3)
diff = A.unsqueeze(1) - B.unsqueeze(0)      # (10, 7, 3)
dist = torch.sqrt((diff ** 2).sum(dim=2))   # (10, 7) — distance matrix

# ── Example 4: outer product via broadcasting ───────────────────
a = torch.tensor([1., 2., 3.])    # (3,)
b = torch.tensor([10., 20.])      # (2,)
outer = a.unsqueeze(1) * b.unsqueeze(0)   # (3,1) * (1,2) → (3,2)
# [[10,20],[20,40],[30,60]]

7.3 Common Broadcasting Mistakes

a = torch.randn(3, 4)
b = torch.randn(4, 3)

# This will FAIL — shapes are not broadcast-compatible in the wrong orientation
# a + b   # RuntimeError!

# Fix 1: Transpose one
a + b.T   # (3,4) + (3,4) ✓

# Fix 2: Explicit reshape
a + b.reshape(3, 4)  # if that's what you meant

# Mistake: missing unsqueeze
col_mean = a.mean(dim=1)   # (3,) — mean of each ROW
# a - col_mean   # tries (3,4) - (3,) → broadcast makes (3,) → (1,3) → ERROR
# Fix: keep dimension
col_mean = a.mean(dim=1, keepdim=True)   # (3, 1)
a - col_mean    # (3,4) - (3,1) → (3,4) ✓ subtracts row mean from each row

✏️ Practice Problem 7

Problem: You have embeddings E of shape (vocab_size=1000, embed_dim=64) and a query vector q of shape (64,). Compute the dot product (similarity score) of the query against every embedding, producing a score vector of shape (1000,).

Solution:

E = torch.randn(1000, 64)
q = torch.randn(64)

# Method 1: Matrix-vector multiply
scores = E @ q    # (1000, 64) @ (64,) → (1000,)

# Method 2: Explicit sum reduction (broadcasting)
scores = (E * q).sum(dim=1)  # (1000,64)*(64,) broadcast → (1000,64) → sum → (1000,)

# Method 3: Using einsum (elegant and explicit)
scores = torch.einsum('vd,d->v', E, q)   # v=vocab, d=dim

# All three produce identical results
# Now find the top-5 most similar vocabulary items
topk_scores, topk_indices = torch.topk(scores, k=5)
print(topk_scores)    # 5 highest scores
print(topk_indices)   # their indices in the vocabulary

8. Matrix Multiplication — The Core Operation

This is the central operation of all linear algebra in PyTorch. Know every variant cold.

8.1 The Full Operator Taxonomy

A = torch.randn(3, 4)
B = torch.randn(4, 5)

# ── 2D matrix multiplication ────────────────────────────────────
C = torch.mm(A, B)           # (3,5) — 2D ONLY, most explicit
C = torch.matmul(A, B)       # (3,5) — works for any ndim
C = A @ B                    # (3,5) — operator alias for matmul

# ── Matrix-vector product ───────────────────────────────────────
v = torch.randn(4)
torch.mv(A, v)               # (3,) — matrix × vector, 2D only
A @ v                        # (3,) — matmul auto-handles (4,) → column vec

# ── Vector dot product (1D only) ────────────────────────────────
a = torch.randn(4)
b = torch.randn(4)
torch.dot(a, b)              # scalar — ONLY for 1D vectors
a @ b                        # scalar — same

# ── Outer product ───────────────────────────────────────────────
torch.outer(a, b)            # (4, 4)
a.unsqueeze(1) @ b.unsqueeze(0)  # same: (4,1) @ (1,4) → (4,4)

# ── matmul is the universal operator ───────────────────────────
# 1D × 1D   → scalar (dot product)
# 2D × 2D   → matrix multiply
# 2D × 1D   → matrix-vector (treats 1D as column vector, result is 1D)
# 1D × 2D   → vector-matrix (treats 1D as row vector, result is 1D)

8.2 matmul Broadcasting Rules

torch.matmul (and @) batch-broadcasts when tensors have more than 2 dimensions. The last two dimensions are the matrix dimensions; all leading dimensions are broadcast.

# Batch of matrices × single matrix
bA = torch.randn(10, 3, 4)   # 10 matrices of shape (3,4)
B  = torch.randn(4, 5)       #  1 matrix  of shape (4,5)
torch.matmul(bA, B)          # (10,3,5) — B broadcast to each of 10 batches

# Batch × Batch (broadcast)
bA = torch.randn(10, 3, 4)
bB = torch.randn(10, 4, 5)
torch.matmul(bA, bB)          # (10,3,5) — one matmul per batch element

# Higher-rank broadcasting
bA = torch.randn(2, 10, 3, 4)
bB = torch.randn(   10, 4, 5)   # implicit (1,10,...) prepended
torch.matmul(bA, bB)            # (2,10,3,5) — broadcast over dim 0

8.3 Understanding the Shapes

The most common source of bugs is shape confusion. Develop this instinct:

(M, K) × (K, N) → (M, N)
      ^   ^
      These must match — the "inner dimensions"

For batched: (B, M, K) × (B, K, N) → (B, M, N)
             (*, M, K) × (*, K, N) → (*, M, N)
             (all leading dims broadcast)

# Worked example: multi-head attention shapes
# Q: (batch, seq_len, d_k)
# K: (batch, seq_len, d_k)
# Attention scores: (batch, seq_len, seq_len)

B, S, d_k = 4, 16, 32
Q = torch.randn(B, S, d_k)
K = torch.randn(B, S, d_k)

# Q @ K^T — K must be transposed to (B, d_k, S)
scores = Q @ K.transpose(-2, -1)   # (4, 16, 16)
# Scale by sqrt(d_k) for numerical stability
scores = scores / (d_k ** 0.5)

8.4 Performance Tips

# ── Use @ or matmul, NOT manual loops ───────────────────────────
A = torch.randn(1000, 1000)
B = torch.randn(1000, 1000)

# SLOW — pure Python loop
result = torch.zeros(1000, 1000)
for i in range(1000):
    result[i] = (A[i].unsqueeze(0) @ B).squeeze()

# FAST — single BLAS call
result = A @ B    # milliseconds vs seconds

# ── Reuse transposed tensors ─────────────────────────────────────
# If you need A.T multiple times, store it
At = A.T.contiguous()   # make contiguous once
for _ in range(100):
    result = At @ B     # faster than (A.T @ B) * 100 times

# ── Choose dtype based on precision needs ───────────────────────
A16 = A.half()   # float16 — 2x faster matmul on GPU, 2x less memory
B16 = B.half()
result = A16 @ B16   # fast but less precise

✏️ Practice Problem 8

Problem: Implement linear regression prediction from scratch using only tensor operations:

Input X has shape (N=100, features=5)
Weights W has shape (5,) and bias b is a scalar
Compute predictions y_hat of shape (N,)
Compute Mean Squared Error loss as a scalar

Solution:

torch.manual_seed(42)
N, F = 100, 5

X = torch.randn(N, F)       # design matrix
W = torch.randn(F)           # weight vector
b = torch.tensor(0.5)        # bias
y = torch.randn(N)           # true labels

# Predictions: X @ W + b
y_hat = X @ W + b            # (100,5)@(5,) → (100,) + scalar → (100,)

# MSE: mean over all samples
residuals = y_hat - y        # (100,)
mse = (residuals ** 2).mean()   # scalar

# Equivalent:
mse = torch.nn.functional.mse_loss(y_hat, y)

print(f"y_hat shape: {y_hat.shape}")   # torch.Size([100])
print(f"MSE: {mse.item():.4f}")

9. Batched Matrix Operations

When you have collections of matrices to operate on simultaneously, batched operations are critical.

9.1 bmm vs matmul

# bmm — STRICTLY 3D (batch, M, K) × (batch, K, N)
bA = torch.randn(8, 3, 4)
bB = torch.randn(8, 4, 5)
torch.bmm(bA, bB)          # (8, 3, 5) ✓

# bmm does NOT broadcast — all three must have same batch size
bC = torch.randn(1, 4, 5)
# torch.bmm(bA, bC)        # ERROR if batch dim != 8

# matmul — handles broadcast over any leading dims
torch.matmul(bA, bC)       # (8, 3, 5) — bC broadcast to 8 batches ✓
torch.matmul(bA, bB)       # (8, 3, 5) ✓ — works like bmm when 3D

9.2 Batched Applications

# ── Batched covariance matrices ──────────────────────────────────
# Given B batches of N samples with D features: (B, N, D)
# Compute covariance: (B, D, D)
data = torch.randn(4, 100, 8)   # 4 batches, 100 samples, 8 features

mean = data.mean(dim=1, keepdim=True)   # (4, 1, 8)
centered = data - mean                   # (4, 100, 8)

# Covariance: (1/N) * X^T X
# centered.transpose(-2,-1) : (4, 8, 100)
# centered                   : (4, 100, 8)
cov = torch.bmm(centered.transpose(-2, -1), centered) / 100   # (4, 8, 8)

# ── Applying different transforms to different batches ───────────
transforms = torch.randn(8, 4, 4)   # 8 different 4×4 transform matrices
points     = torch.randn(8, 4, 100)  # 8 batches of 100 4D points
transformed = torch.bmm(transforms, points)   # (8, 4, 100)

# ── Solving multiple linear systems simultaneously ───────────────
# Ax = b for A: (B, N, N), b: (B, N, K) → solve for x: (B, N, K)
A = torch.randn(5, 4, 4)
A = A @ A.transpose(-2,-1) + torch.eye(4)  # make positive definite
b = torch.randn(5, 4, 3)
x = torch.linalg.solve(A, b)   # (5, 4, 3) — solves all 5 systems

✏️ Practice Problem 9

Problem: You have a batch of 16 square matrices, each 6×6 (shape (16, 6, 6)). Compute the trace (sum of diagonal elements) of each matrix.

Solution:

batch = torch.randn(16, 6, 6)

# Method 1: diagonal + sum
# torch.diagonal returns shape (6, 16) by default — need dim1, dim2
diags = torch.diagonal(batch, dim1=-2, dim2=-1)   # (16, 6)
traces = diags.sum(dim=-1)                          # (16,)

# Method 2: einsum — trace is contraction of diagonal indices
traces = torch.einsum('bii->b', batch)   # b=batch, i=diagonal index

# Method 3: Using linalg (most readable)
traces = torch.vmap(torch.trace)(batch)  # PyTorch 2.0+ vmap

print(traces.shape)   # torch.Size([16])
print(traces[:3])

10. Einstein Summation (einsum)

torch.einsum is the most expressive way to write tensor contractions. Once you understand the notation, it replaces nested loops, transposes, and complex matmul chains with a single readable string.

10.1 The Einsum Grammar

The notation is: 'input_subscripts -> output_subscripts'

Each letter represents one dimension
Repeated letters on the left are contracted (summed over)
Letters appearing only in the output are kept
Letters in the input but missing from the output are summed away

'ij,jk->ik'
 ^^  ^^  ^^
 A   B   C

i → output dimension (rows of A / rows of C)
j → contracted dimension (must match: cols of A == rows of B)
k → output dimension (cols of B / cols of C)
Result: C[i,k] = Σ_j A[i,j] * B[j,k]

10.2 Einsum for Common Operations

import torch

A = torch.randn(3, 4)
B = torch.randn(4, 5)
a = torch.randn(3)
b = torch.randn(3)
v = torch.randn(4)

# ── Linear Algebra ──────────────────────────────────────────────
torch.einsum('ij,jk->ik', A, B)     # (3,5) — matrix multiply
torch.einsum('ij,j->i', A, v)       # (3,)  — matrix-vector multiply
torch.einsum('i,i->', a, b)         # scalar — dot product
torch.einsum('i,j->ij', a, b)       # (3,3) — outer product
torch.einsum('ij->ji', A)           # (4,3) — transpose
torch.einsum('ii->', A[:min(A.shape):])  # scalar — trace (square submatrix)
torch.einsum('ij->', A)             # scalar — sum all elements
torch.einsum('ij->i', A)            # (3,)  — row sums
torch.einsum('ij->j', A)            # (4,)  — column sums

# ── Element-wise / Hadamard ─────────────────────────────────────
C = torch.randn(3, 4)
torch.einsum('ij,ij->ij', A, C)    # (3,4) — element-wise multiply (A*C)
torch.einsum('ij,ij->', A, C)      # scalar — Frobenius inner product (A*C).sum()

# ── Batched operations ──────────────────────────────────────────
bA = torch.randn(8, 3, 4)
bB = torch.randn(8, 4, 5)
torch.einsum('bij,bjk->bik', bA, bB)   # (8,3,5) — batched matmul

# ── Diagonal extraction ─────────────────────────────────────────
sq = torch.randn(4, 4)
torch.einsum('ii->i', sq)              # (4,) — diagonal elements

# ── Bilinear form: x^T A y ─────────────────────────────────────
x, y = torch.randn(4), torch.randn(4)
torch.einsum('i,ij,j->', x, sq, y)    # scalar

# ── Multi-head attention score computation ─────────────────────
# Q: (batch, heads, seq, d_k)  K: (batch, heads, seq, d_k)
B, H, S, d = 2, 4, 16, 32
Q = torch.randn(B, H, S, d)
K = torch.randn(B, H, S, d)
# Attention scores: (batch, heads, seq_q, seq_k)
scores = torch.einsum('bhqd,bhkd->bhqk', Q, K)   # (2,4,16,16)

10.3 Understanding Contraction Step by Step

# Let's trace 'ij,jk->ik' by hand for small tensors
A = torch.tensor([[1., 2.],
                   [3., 4.]])   # (2,2) — i=2, j=2
B = torch.tensor([[5., 6., 7.],
                   [8., 9., 10.]])  # (2,3) — j=2, k=3

result = torch.einsum('ij,jk->ik', A, B)
# result[0,0] = A[0,0]*B[0,0] + A[0,1]*B[1,0] = 1*5 + 2*8 = 21
# result[0,1] = A[0,0]*B[0,1] + A[0,1]*B[1,1] = 1*6 + 2*9 = 24
# result[0,2] = A[0,0]*B[0,2] + A[0,1]*B[1,2] = 1*7 + 2*10 = 27
# result[1,0] = A[1,0]*B[0,0] + A[1,1]*B[1,0] = 3*5 + 4*8 = 47
print(result)
# tensor([[21., 24., 27.],
#          [47., 54., 61.]])

✏️ Practice Problem 10

Problem: Using einsum, implement:

Batch dot products: given A (shape (B, D)) and B (shape (B, D)), compute the dot product of each corresponding pair → shape (B,)
Hadamard product summed: element-wise product of two (M, N) matrices, then sum → scalar
Tensor contraction: for T of shape (A, B, C) and M of shape (C, D), contract the last dim → shape (A, B, D)

Solution:

B_size, D = 8, 16
A_t = torch.randn(B_size, D)
B_t = torch.randn(B_size, D)
M_t = torch.randn(3, 4)
N_t = torch.randn(3, 4)
T   = torch.randn(5, 6, 7)
M   = torch.randn(7, 9)

# 1. Batch dot products
batch_dots = torch.einsum('bd,bd->b', A_t, B_t)
print(batch_dots.shape)   # (8,)

# Verify against loop
for i in range(B_size):
    assert torch.allclose(batch_dots[i], torch.dot(A_t[i], B_t[i]))

# 2. Hadamard product summed (Frobenius inner product)
fro = torch.einsum('ij,ij->', M_t, N_t)
print(fro.shape)   # scalar = torch.Size([])

# 3. Tensor-matrix contraction over last dim
result = torch.einsum('abc,cd->abd', T, M)
print(result.shape)   # (5, 6, 9)

# Verify with matmul (equivalent)
result2 = (T.reshape(-1, 7) @ M).reshape(5, 6, 9)
print(torch.allclose(result, result2))   # True

11. Linear Algebra — linalg Module

torch.linalg is the modern (PyTorch 1.9+) home for all linear algebra operations. Prefer it over older scattered functions.

11.1 Norms

t = torch.tensor([[3., 4.], [1., 2.]])

# Vector norms (when applied to a 1D tensor or with dim specified)
torch.linalg.norm(torch.tensor([3., 4.]))      # L2 norm: 5.0  (√(9+16))
torch.linalg.norm(t, ord=2)                    # spectral norm (max singular value)
torch.linalg.norm(t, ord='fro')               # Frobenius norm: √(9+16+1+4) = √30
torch.linalg.norm(t, ord=1)                    # max column sum
torch.linalg.norm(t, ord=float('inf'))         # max row sum

# Per-row or per-column norms (using dim parameter)
torch.linalg.norm(t, dim=1)       # L2 norm of each row: [5.0, 2.236]
torch.linalg.norm(t, dim=0)       # L2 norm of each column: [3.162, 4.472]

# L1 norm of rows (used in normalisation)
torch.linalg.norm(t, ord=1, dim=1)   # sum of abs values per row

# Normalise rows to unit L2 norm
normed = t / torch.linalg.norm(t, dim=1, keepdim=True)
print(torch.linalg.norm(normed, dim=1))   # [1., 1.] — unit vectors

11.2 Determinant and Inverse

A = torch.tensor([[4., 7.], [2., 6.]], dtype=torch.float32)

# Determinant
det = torch.linalg.det(A)            # tensor(10.)  (4*6 - 7*2)
logdet = torch.linalg.slogdet(A)     # (sign, log|det|) — numerically stable

# Inverse — use only when you need the matrix explicitly
A_inv = torch.linalg.inv(A)
print(A @ A_inv)                      # should be close to identity

# BETTER: when solving Ax = b, never invert!
b = torch.tensor([1., 0.])
x = torch.linalg.solve(A, b)          # more stable and faster than inv(A) @ b

# Batched versions work the same way
batch_A = torch.randn(10, 4, 4)
batch_b = torch.randn(10, 4, 2)
batch_A = batch_A @ batch_A.transpose(-2,-1) + torch.eye(4)  # positive definite
x = torch.linalg.solve(batch_A, batch_b)   # (10, 4, 2)

11.3 Eigendecomposition

# For symmetric/Hermitian matrices (more stable)
A_sym = torch.randn(4, 4)
A_sym = A_sym @ A_sym.T            # A_sym is now symmetric positive semi-definite

eigenvalues = torch.linalg.eigvalsh(A_sym)          # real eigenvalues (sorted!)
eigenvalues, eigenvectors = torch.linalg.eigh(A_sym) # both

# For general matrices
eigenvalues_complex = torch.linalg.eigvals(A_sym)    # may return complex
eigenvalues_c, eigenvectors_c = torch.linalg.eig(A_sym)

# Reconstruction: A = V @ diag(Λ) @ V^{-1}
# For symmetric: A = V @ diag(Λ) @ V.T  (V is orthogonal)
A_reconstructed = eigenvectors @ torch.diag(eigenvalues) @ eigenvectors.T
print(torch.allclose(A_sym, A_reconstructed, atol=1e-5))   # True

11.4 Singular Value Decomposition (SVD)

SVD is one of the most important decompositions. It works for any matrix (not just square), and is the foundation of PCA, low-rank approximation, and pseudo-inverse.

A = U @ diag(S) @ Vh

A:  (M, N)  — the original matrix
U:  (M, M)  — left singular vectors (orthogonal columns)
S:  (min(M,N),) — singular values (non-negative, sorted descending)
Vh: (N, N)  — right singular vectors (Vh is V^H, the conjugate transpose of V)

A = torch.randn(5, 3)   # (5, 3)

# Full SVD
U, S, Vh = torch.linalg.svd(A)
print(U.shape)   # (5, 5)
print(S.shape)   # (3,) — min(5,3) = 3
print(Vh.shape)  # (3, 3)

# Reconstruct A
A_recon = U[:, :3] @ torch.diag(S) @ Vh    # need only first 3 cols of U
print(torch.allclose(A, A_recon, atol=1e-5))  # True

# Truncated SVD (economy/thin) — more efficient
U, S, Vh = torch.linalg.svd(A, full_matrices=False)
print(U.shape)   # (5, 3) — economy
print(S.shape)   # (3,)
print(Vh.shape)  # (3, 3)

# Reconstruct directly
A_recon = U @ torch.diag(S) @ Vh
print(torch.allclose(A, A_recon, atol=1e-5))  # True

# ── Low-rank approximation (keep top k singular values) ───────────
def low_rank_approx(A, k):
    U, S, Vh = torch.linalg.svd(A, full_matrices=False)
    return U[:, :k] @ torch.diag(S[:k]) @ Vh[:k, :]

rank2_approx = low_rank_approx(A, k=2)
print("Frobenius error:", torch.linalg.norm(A - rank2_approx, ord='fro'))

# ── Pseudo-inverse (for non-square or singular matrices) ──────────
A_pinv = torch.linalg.pinv(A)   # (3, 5) — right pseudo-inverse
print(A_pinv.shape)
# Satisfies: A @ A_pinv @ A ≈ A

11.5 QR and Cholesky Decompositions

# QR Decomposition: A = Q @ R
# Q: orthogonal, R: upper triangular
A = torch.randn(5, 3)
Q, R = torch.linalg.qr(A)
print(Q.shape)  # (5, 3) — economy QR by default
print(R.shape)  # (3, 3)
print(torch.allclose(A, Q @ R, atol=1e-5))   # True

# Q is orthogonal: Q.T @ Q ≈ I
print(torch.allclose(Q.T @ Q, torch.eye(3), atol=1e-5))   # True

# Cholesky: A = L @ L.T (for positive definite A)
A = torch.randn(4, 4)
A = A @ A.T + torch.eye(4)   # make positive definite
L = torch.linalg.cholesky(A)
print(torch.allclose(L @ L.T, A, atol=1e-5))   # True

✏️ Practice Problem 11 — PCA from Scratch

Problem: Implement PCA (Principal Component Analysis) using SVD to project a data matrix X of shape (200, 10) down to 3 principal components.

Solution:

torch.manual_seed(0)
X = torch.randn(200, 10)   # 200 samples, 10 features

# Step 1: Center the data (subtract column means)
mean = X.mean(dim=0)           # (10,)
X_centered = X - mean          # (200, 10) — broadcast

# Step 2: Compute SVD of the centred data
# Covariance structure is encoded in the right singular vectors Vh
U, S, Vh = torch.linalg.svd(X_centered, full_matrices=False)
# U:  (200, 10)
# S:  (10,)      — singular values (√eigenvalues of covariance * N)
# Vh: (10, 10)   — rows are principal components

# Step 3: Project onto top-3 principal components
# Principal components are the ROWS of Vh (right singular vectors)
components = Vh[:3, :]    # (3, 10) — top 3 PCs
X_projected = X_centered @ components.T   # (200, 10) @ (10, 3) → (200, 3)

print(f"Original:  {X.shape}")        # (200, 10)
print(f"Projected: {X_projected.shape}")  # (200, 3)

# Explained variance ratio
total_var = (S ** 2).sum()
explained = (S[:3] ** 2) / total_var
print(f"Variance explained: {explained.tolist()}")
print(f"Cumulative: {explained.cumsum(0).tolist()}")

# Reconstruction from top-3 (lossy)
X_reconstructed = X_projected @ components + mean   # (200, 3) @ (3, 10) + (10,)
reconstruction_error = torch.linalg.norm(X - X_reconstructed, ord='fro')
print(f"Reconstruction error: {reconstruction_error:.4f}")

12. Advanced Indexing — gather & scatter

These operations are fundamental for NLP (token embedding lookups, masked attention), RL (action selection), and any problem requiring non-uniform selection or writing.

12.1 torch.gather — Selective Read

gather collects values from input at specified index positions along a dimension.

output[i][j][k] = input[i][index[i][j][k]][k]   (for dim=1)

Crucially: output has the same shape as index.

t = torch.tensor([[10., 20., 30.],
                   [40., 50., 60.],
                   [70., 80., 90.]])   # (3, 3)

# For each row, select specific columns
idx = torch.tensor([[0, 2],     # row 0: pick columns 0 and 2
                    [1, 1],     # row 1: pick columns 1 and 1 (same)
                    [2, 0]])    # row 2: pick columns 2 and 0
result = torch.gather(t, dim=1, index=idx)
# result[0] = [t[0,0], t[0,2]] = [10, 30]
# result[1] = [t[1,1], t[1,1]] = [50, 50]
# result[2] = [t[2,2], t[2,0]] = [90, 70]
print(result)
# tensor([[10., 30.],
#          [50., 50.],
#          [90., 70.]])

12.2 Practical gather — One-Hot, Token Lookup

# ── Token embedding lookup (what nn.Embedding does internally) ─
vocab_size, embed_dim = 1000, 64
embeddings = torch.randn(vocab_size, embed_dim)   # (V, D)
token_ids  = torch.randint(0, vocab_size, (8, 16)) # (B, S) batch of token sequences

# Method 1: fancy indexing (simple, works)
embedded = embeddings[token_ids]   # (8, 16, 64)

# ── Gathering logits for specific classes ──────────────────────
# batch_size=8, num_classes=10
logits  = torch.randn(8, 10)
targets = torch.tensor([3, 7, 1, 0, 9, 4, 2, 6])   # one target per sample

# Gather the logit corresponding to the true class
# index must match logits.ndim — need shape (8, 1)
idx     = targets.unsqueeze(1)           # (8, 1)
correct_logits = torch.gather(logits, dim=1, index=idx)  # (8, 1)
correct_logits = correct_logits.squeeze(1)               # (8,)

# ── NLL loss from scratch ──────────────────────────────────────
log_probs = torch.log_softmax(logits, dim=1)   # (8, 10)
nll = -torch.gather(log_probs, 1, targets.unsqueeze(1)).squeeze()  # (8,)
loss = nll.mean()   # scalar

12.3 torch.scatter_ and torch.scatter — Selective Write

scatter_ is the inverse of gather: instead of reading from specific positions, it writes to them. Note the trailing underscore — it's in-place by default.

output[i][index[i][j]][k] = src[i][j][k]   (for dim=1)

# ── One-hot encoding ──────────────────────────────────────────
batch_size, num_classes = 8, 5
labels = torch.tensor([0, 3, 1, 4, 2, 0, 3, 1])   # (8,)

one_hot = torch.zeros(batch_size, num_classes)      # (8, 5)
one_hot.scatter_(1, labels.unsqueeze(1), 1.0)       # write 1.0 at label positions
print(one_hot)
# tensor([[1,0,0,0,0],
#          [0,0,0,1,0],
#          [0,1,0,0,0], ...])

# ── Scatter add (accumulate) ──────────────────────────────────
# Count occurrences of each token in a batch of sequences
vocab_size = 10
tokens = torch.tensor([3, 1, 3, 5, 1, 2, 3])   # token IDs
counts = torch.zeros(vocab_size)
counts.scatter_add_(0, tokens, torch.ones_like(tokens, dtype=torch.float))
print(counts)   # [0,2,1,3,0,1,0,0,0,0] — token 3 appears 3 times

# ── Functional scatter (non-in-place) ─────────────────────────
src = torch.ones(2, 3)
idx = torch.tensor([[0, 1, 2], [0, 1, 4]])
out = torch.zeros(2, 5)
out.scatter_(1, idx, src)   # writes src values at idx positions

12.4 torch.index_select, torch.index_put

t = torch.randn(5, 4)

# index_select — select rows or columns by index tensor
selected_rows = torch.index_select(t, 0, torch.tensor([0, 2, 4]))  # (3, 4)
selected_cols = torch.index_select(t, 1, torch.tensor([1, 3]))     # (5, 2)

# index_put — batch assignment
indices = (torch.tensor([0, 2, 4]), torch.tensor([1, 3, 0]))
values  = torch.tensor([99., 99., 99.])
t.index_put_(indices, values)   # t[0,1]=99, t[2,3]=99, t[4,0]=99

✏️ Practice Problem 12

Problem: Implement a "masked softmax" without any loops:

Input: logits of shape (B=4, S=8) — attention logits
mask of shape (B=4, S=8) — boolean, True = position to IGNORE
Compute softmax over logits, setting ignored positions to 0 in the output

Solution:

B, S = 4, 8
logits = torch.randn(B, S)
mask   = torch.zeros(B, S, dtype=torch.bool)
mask[:, -2:] = True   # ignore last 2 positions in every sequence

# Fill ignored positions with -infinity BEFORE softmax
# This makes them contribute 0 after exp(x) / sum(exp(x))
masked_logits = logits.masked_fill(mask, float('-inf'))  # (B, S)

# Numerically stable softmax
attention = torch.softmax(masked_logits, dim=-1)   # (B, S)

# Verify: masked positions are 0
print(attention[:, -2:])   # should be [[0,0],[0,0],[0,0],[0,0]]

# Verify: non-masked rows sum to 1
print(attention.sum(dim=-1))   # [1., 1., 1., 1.]

# Alternative: torch.where approach
NEG_INF = torch.full_like(logits, float('-inf'))
masked_logits2 = torch.where(mask, NEG_INF, logits)
attention2 = torch.softmax(masked_logits2, dim=-1)
print(torch.allclose(attention, attention2))   # True

13. Memory Layout, Strides & Contiguity

Understanding this makes you dangerous in code reviews and debugging sessions.

13.1 How Tensors Are Stored

A tensor's data is a flat 1D array in memory. The strides tell PyTorch how many elements to skip to advance by 1 in each dimension.

t = torch.arange(12).reshape(3, 4)

Memory:  [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
Shape:   (3, 4)
Strides: (4, 1)

t[i, j] is at memory offset: i * 4 + j * 1
t[0,0]=0, t[0,1]=1, t[1,0]=4, t[2,3]=11 ✓

t = torch.arange(12).reshape(3, 4)
print(t.stride())    # (4, 1) — C-contiguous (row-major)

t_T = t.T           # transpose — SWAP strides, no copy
print(t_T.stride())  # (1, 4) — Fortran-contiguous (column-major)
print(t_T.is_contiguous())   # False

# Strides for a sliced tensor
s = t[::2, ::2]     # every other row and column
print(s.stride())    # (8, 2) — skipping over elements

# Make contiguous (allocates new memory, rewrites data)
s_c = s.contiguous()
print(s_c.stride())  # (2, 1) — back to C-contiguous

13.2 Why Contiguity Matters

t = torch.randn(4, 4)
t_T = t.T

# Some operations require contiguous memory
# t_T.view(16)   # RuntimeError: non-contiguous
t_T.reshape(16)  # OK — reshape handles non-contiguous with an internal copy

# Storage_offset: where in the underlying storage does the tensor start
s = t[1:]          # starts at element 4 in storage
print(s.storage_offset())   # 4

# Checking if two tensors share storage
print(t.storage().data_ptr() == t_T.storage().data_ptr())  # True — shared!
print(t.storage().data_ptr() == t_T.contiguous().storage().data_ptr())  # False — copy

13.3 Performance Implications

# Contiguous tensors are cache-friendly — sequential memory access
t = torch.randn(1000, 1000)

# Fast: row-major access matches C-contiguous layout
row_sums = t.sum(dim=1)    # iterates along contiguous dimension last

# Operations that force contiguous copies are expensive
# Avoid in tight loops:
t_T = t.T.contiguous()    # forces copy
# Do instead: plan your layout so you don't need to transpose repeatedly

# float16 vs float32 memory
t32 = torch.randn(1000, 1000)           # 4 MB
t16 = t32.half()                         # 2 MB — half the memory
t16 = t32.to(torch.float16)             # equivalent

# In-place ops save memory (no intermediate tensor)
t.add_(1.0)     # modifies t, no new allocation
t.relu_()       # same

14. Practice Problems with Full Solutions

These problems are calibrated to the style and difficulty of your assessment.

Problem A — Shape Surgery

Problem: Given x of shape (B=4, C=3, H=8, W=8) (a batch of images), produce a tensor where each image's pixels are flattened into a row, resulting in shape (4, 192).

x = torch.randn(4, 3, 8, 8)

# Method 1: reshape
out = x.reshape(4, -1)         # (4, 192)

# Method 2: flatten (cleaner API)
out = x.flatten(start_dim=1)   # flatten all dims except batch

# Method 3: view (only if contiguous)
out = x.contiguous().view(4, -1)

print(out.shape)   # torch.Size([4, 192])

Problem B — Efficient Softmax from Scratch

Problem: Implement a numerically stable softmax (without using torch.softmax). For input x of shape (B, C), compute softmax over the class dimension.

def stable_softmax(x):
    # Subtract max for numerical stability: softmax(x) = softmax(x - c)
    x_max = x.max(dim=1, keepdim=True).values   # (B, 1)
    x_shifted = x - x_max                        # (B, C) — broadcast

    exp_x = torch.exp(x_shifted)                 # (B, C)
    sum_exp = exp_x.sum(dim=1, keepdim=True)      # (B, 1)
    return exp_x / sum_exp                        # (B, C) — broadcast

x = torch.randn(4, 10)
out = stable_softmax(x)

# Verify: rows sum to 1
print(out.sum(dim=1))   # [1., 1., 1., 1.]

# Verify: all values non-negative
print((out >= 0).all())  # True

# Verify: matches PyTorch's softmax
print(torch.allclose(out, torch.softmax(x, dim=1), atol=1e-6))  # True

Problem C — Pairwise Cosine Similarity Matrix

Problem: Given a matrix A of shape (N, D) where each row is a vector, compute the full (N, N) cosine similarity matrix without any Python loops.

def cosine_similarity_matrix(A):
    # Normalise each row to unit L2 norm
    norms = torch.linalg.norm(A, dim=1, keepdim=True)   # (N, 1)
    norms = torch.clamp(norms, min=1e-8)                  # avoid division by zero
    A_norm = A / norms                                     # (N, D)

    # Cosine similarity = dot product of unit vectors = A_norm @ A_norm.T
    return A_norm @ A_norm.T   # (N, N)

N, D = 10, 16
A = torch.randn(N, D)
sim = cosine_similarity_matrix(A)

print(sim.shape)             # (10, 10)
print(sim.diagonal())        # all 1.0 — each vector is identical to itself
print(sim.min(), sim.max())  # values in [-1, 1]

# Verify diagonal is exactly 1.0
print(torch.allclose(sim.diagonal(), torch.ones(N), atol=1e-5))  # True

Problem D — Sliding Window Sum (Strided Trick)

Problem: Given a 1D signal x of length N=12, compute the sum of every window of size k=4 without a Python loop, producing output of length N-k+1=9.

x = torch.arange(1., 13.)   # [1,2,3,4,5,6,7,8,9,10,11,12]
k = 4

# Method 1: unfold + sum (most elegant)
windows = x.unfold(dimension=0, size=k, step=1)   # (9, 4)
window_sums = windows.sum(dim=1)                   # (9,)
print(window_sums)
# [10, 14, 18, 22, 26, 30, 34, 38, 42]

# Method 2: cumulative sum trick — O(N) not O(N*k)
cumsum = torch.cat([torch.zeros(1), x.cumsum(0)])   # (13,)
window_sums2 = cumsum[k:] - cumsum[:-k]              # (9,)
print(torch.allclose(window_sums, window_sums2))    # True

# What does unfold give us?
print(windows)
# tensor([[ 1,  2,  3,  4],
#          [ 2,  3,  4,  5],
#          ...
#          [ 9, 10, 11, 12]])

Problem E — Masked Fill and Causal Attention Mask

Problem: Create a causal (lower-triangular) mask for a sequence of length S=6 and apply it to an attention score matrix, setting future positions to -inf so softmax ignores them.

S = 6

# Create causal mask — True where we should MASK (upper triangle, excluding diagonal)
causal_mask = torch.triu(torch.ones(S, S, dtype=torch.bool), diagonal=1)
print(causal_mask.int())
# [[0,1,1,1,1,1],
#  [0,0,1,1,1,1],
#  [0,0,0,1,1,1],
#  [0,0,0,0,1,1],
#  [0,0,0,0,0,1],
#  [0,0,0,0,0,0]]

# Apply to attention scores
attention_scores = torch.randn(2, 4, S, S)   # (batch, heads, S, S)

# Broadcast mask: (S, S) → applied to all batches and heads automatically
masked_scores = attention_scores.masked_fill(causal_mask, float('-inf'))

# Compute attention weights
attention_weights = torch.softmax(masked_scores, dim=-1)   # (2,4,S,S)

# Verify: upper triangle is 0 (softmax of -inf)
print(attention_weights[0, 0])   # lower triangular structure
print((attention_weights[0, 0] > 0).int())   # should be lower triangular

Problem F — Batch Matrix Inversion with Validation

Problem: Given a batch of B=5 random 4×4 matrices, make them invertible (positive definite), compute their inverses, and verify A @ A_inv ≈ I for each.

B, N = 5, 4

# Generate random positive definite matrices
A_raw = torch.randn(B, N, N)
A = torch.bmm(A_raw, A_raw.transpose(-2, -1)) + torch.eye(N)   # (B,N,N)

# Compute batch inverse
A_inv = torch.linalg.inv(A)   # (B, N, N)

# Verify: A @ A_inv should be identity for each batch element
product = torch.bmm(A, A_inv)   # (B, N, N)
identity = torch.eye(N).expand(B, N, N)

max_error = (product - identity).abs().max()
print(f"Max deviation from identity: {max_error:.2e}")   # should be ~1e-6

# Better approach for Ax=b: don't invert, just solve
b = torch.randn(B, N, 2)
x = torch.linalg.solve(A, b)   # (B, N, 2) — numerically superior
print(torch.allclose(torch.bmm(A, x), b, atol=1e-5))   # True

Problem G — Custom Distance Matrix (Assessment Style)

Problem: Implement L2 (Euclidean) pairwise distance between two sets of points without any Python loops:

A: shape (N=50, D=8)
B: shape (M=30, D=8)
Output: (N, M) where out[i,j] = ||A[i] - B[j]||_2

def pairwise_l2(A, B):
    # ||a - b||^2 = ||a||^2 + ||b||^2 - 2 * a.b
    # This avoids the (N, M, D) intermediate tensor from naive subtraction

    A_sq = (A ** 2).sum(dim=1, keepdim=True)   # (N, 1)
    B_sq = (B ** 2).sum(dim=1)                  # (M,) → need (1, M)

    # (N,1) + (1,M) - 2*(N,M) → (N,M)
    sq_dist = A_sq + B_sq.unsqueeze(0) - 2 * (A @ B.T)

    # Clamp to avoid negative values from floating point errors near 0
    return torch.sqrt(torch.clamp(sq_dist, min=0.0))

N, M, D = 50, 30, 8
A = torch.randn(N, D)
B = torch.randn(M, D)

dist = pairwise_l2(A, B)
print(dist.shape)   # (50, 30)

# Verify one distance manually
i, j = 3, 7
manual = torch.linalg.norm(A[i] - B[j])
print(torch.allclose(dist[i, j], manual, atol=1e-5))   # True

# Find nearest neighbour for each point in A
nn_indices = dist.argmin(dim=1)   # (N,) — index in B closest to each point in A
nn_dists   = dist.min(dim=1).values
print(nn_indices.shape, nn_dists.shape)   # (50,) (50,)

Problem H — Full Einsum Challenge

Problem: Use einsum only (no @, mm, matmul) to:

Compute W @ X.T where W: (4,3), X: (5,3) → (4,5)
Compute the batch outer product of rows of A: (B,D) → (B,D,D)
Compute a weighted sum: weights: (B,S), values: (B,S,D) → (B,D)

W = torch.randn(4, 3)
X = torch.randn(5, 3)
A = torch.randn(8, 6)
B_size, S, D_size = 4, 10, 16
weights = torch.softmax(torch.randn(B_size, S), dim=1)
values  = torch.randn(B_size, S, D_size)

# 1. W @ X.T: (4,3) × (3,5) → (4,5)
result1 = torch.einsum('ij,kj->ik', W, X)   # j is contracted, k appears in output
print(result1.shape)   # (4, 5)
print(torch.allclose(result1, W @ X.T))     # True

# 2. Batch outer product: for each b, outer(A[b], A[b]) → (B,D,D)
result2 = torch.einsum('bi,bj->bij', A, A)   # no contraction — pure outer
print(result2.shape)   # (8, 6, 6)
# result2[b] = outer product of A[b] with itself

# 3. Weighted sum: for each batch, sum_s weight[b,s]*value[b,s,:] → (B,D)
result3 = torch.einsum('bs,bsd->bd', weights, values)
print(result3.shape)   # (4, 16)
# Verify against manual:
result3_manual = (weights.unsqueeze(-1) * values).sum(dim=1)  # (B,1)*(B,S,D).sum(1)
print(torch.allclose(result3, result3_manual, atol=1e-6))   # True

15. Assessment Cheat Sheet

A one-stop reference for the most common operations you'll need in 60 minutes.

Tensor Creation

Operation	Code	Output Shape
From list	`torch.tensor([[1,2],[3,4]])`	`(2,2)`
Zeros	`torch.zeros(3,4)`	`(3,4)`
Identity	`torch.eye(N)`	`(N,N)`
Range	`torch.arange(0, 10, 2)`	`(5,)`
Linspace	`torch.linspace(0, 1, 5)`	`(5,)`
Normal	`torch.randn(B, D)`	`(B,D)`
Like	`torch.zeros_like(x)`	same as `x`

Shape Operations

Operation	Code	Notes
Reshape	`t.reshape(2, -1)`	-1 = infer
View	`t.view(N)`	must be contiguous
Transpose (2D)	`t.T` or `t.t()`	shares memory
Transpose (any)	`t.transpose(1, 2)`	swap two dims
Permute	`t.permute(2,0,1)`	reorder all dims
Add dim	`t.unsqueeze(0)`	insert size-1
Remove dim	`t.squeeze(0)`	remove size-1
Flatten	`t.flatten(1)`	flatten from dim
Cat	`torch.cat([a,b], 0)`	join existing dim
Stack	`torch.stack([a,b], 0)`	create new dim

Indexing

Pattern	Code	Notes
Slice (view)	`t[1:3, :]`	shares memory
Bool mask (copy)	`t[t > 0]`	1D result
Fancy (copy)	`t[[0,2,4]]`	new tensor
Clone	`t.clone()`	independent copy
Masked fill	`t.masked_fill(mask, -inf)`	in-place variant: `masked_fill_`
Where	`torch.where(c, a, b)`	vectorised ternary

Matrix Operations

Operation	Code	Dims	Notes
Matrix multiply	`A @ B`	2D+	universal
2D only	`torch.mm(A, B)`	2D	explicit
Batched	`torch.bmm(A, B)`	3D	no broadcast
Batch broadcast	`torch.matmul(A, B)`	any	broadcasts
Einsum	`torch.einsum('ij,jk->ik', A, B)`	any	most expressive
Outer product	`torch.outer(a, b)`	1D	returns 2D
Dot product	`torch.dot(a, b)`	1D	scalar

Reductions

Operation	Code	Notes
Sum all	`t.sum()`	scalar
Sum axis	`t.sum(dim=1)`	collapses dim 1
Keep dim	`t.sum(dim=1, keepdim=True)`	for broadcasting
Argmax	`t.argmax(dim=1)`	index of max
Topk	`torch.topk(t, k=5)`	values + indices
Norm	`torch.linalg.norm(t, dim=1)`	L2 per row

Linear Algebra

Operation	Code	Notes
Inverse	`torch.linalg.inv(A)`	square only
Solve Ax=b	`torch.linalg.solve(A, b)`	prefer over inv
SVD	`torch.linalg.svd(A, full_matrices=False)`	returns U,S,Vh
Eigenvalues	`torch.linalg.eigvalsh(A)`	symmetric
QR	`torch.linalg.qr(A)`	orthogonal + upper tri
Det	`torch.linalg.det(A)`	scalar
Norm	`torch.linalg.norm(A, ord='fro')`	Frobenius
Pseudo-inv	`torch.linalg.pinv(A)`	for non-square

Advanced Indexing

Operation	Code	Notes
Gather	`torch.gather(t, dim=1, index=idx)`	output = idx.shape
Scatter	`out.scatter_(1, idx, src)`	in-place write
Scatter add	`out.scatter_add_(0, idx, src)`	accumulate
Index select	`t.index_select(0, idx)`	select rows/cols

Common Broadcasting Patterns

# Row-wise operation: subtract column mean
mean = t.mean(dim=0)           # (C,)
t - mean                        # (N,C) - (C,) ✓ (right-aligns)

# Col-wise operation: subtract row mean
mean = t.mean(dim=1, keepdim=True)  # (N,1) — keepdim is CRITICAL
t - mean                             # (N,C) - (N,1) → (N,C) ✓

# 3D broadcast: add bias to each batch
bias = torch.randn(C)           # (C,)
t = torch.randn(B, N, C)
t + bias                        # (B,N,C) + (C,) ✓

# Outer product via unsqueeze
a = torch.randn(M)   # (M,)
b = torch.randn(N)   # (N,)
a.unsqueeze(1) * b.unsqueeze(0)   # (M,1) * (1,N) → (M,N) ✓

Critical Gotchas to Avoid in the Assessment

a * b is element-wise, NOT matrix multiply — use a @ b for matmul
t.T is non-contiguous — if the next op needs contiguity, call .contiguous() first
torch.dot is 1D only — use @ or einsum for higher dimensions
reshape vs view: view fails on non-contiguous tensors, reshape doesn't
Slices are views — mutating a slice mutates the original tensor
keepdim=True in reductions — critical when you need to broadcast the result back
bmm does NOT broadcast — use matmul or @ for broadcasting batch operations
Float/int mixing — always cast before mixed operations to avoid type errors
squeeze() with no args removes ALL size-1 dims — be specific: squeeze(0)
gather output shape = index shape — size the index tensor as the output you want

Happy studying! Focus on the Practice Problems (section 14) — work through each one without looking at the solution first, then check your answer. That's the fastest way to build the pattern recognition you'll need under time pressure.

API Testing in VS Code Using `api.http` (REST Client) — A Practical Guide for Experienced Developers

Ajit Kumar — Wed, 25 Feb 2026 08:07:08 +0000

If you’ve spent years using tools like Postman, you already understand the value of structured API testing, collections, environments, and documentation. But as projects mature, you may start wanting something:

Lightweight
Version-controlled
Close to the codebase
Zero context switching

That’s where the REST Client extension in VS Code with .http files becomes a serious alternative.

For a 4–5 year experienced developer, this isn’t about replacing Postman. It’s about:

Keeping API examples in Git
Reducing tool overhead
Treating API calls as documentation
Making backend repos self-testable

This guide walks you from installation to advanced environment setup, including two major issues most developers hit:

--- vs ### request separation and JSON parsing errors
Why rest-client.environment.json sometimes doesn’t work and how .vscode/settings.json solved it

1️⃣ Why Use `api.http` Instead of Postman?

For experienced developers, the real advantages are:

✅ Version Control Friendly

.http files live in your repository. API examples evolve with the backend.

✅ Lightweight

No separate app. No syncing accounts. No collections export.

✅ Documentation + Testing in One

Your api.http becomes living documentation.

✅ CI-Friendly

You can later convert these into automated tests.

2️⃣ Installing REST Client in VS Code

Open Extensions (Ctrl + Shift + X)
Search for:

REST Client

Install:

REST Client by Huachao Mao

Reload VS Code.

3️⃣ Creating Your First `api.http`

Create a file:

api.http

Let’s use a free public API for demo:

We’ll use:

https://jsonplaceholder.typicode.com

Example 1 — Simple GET Request

### Get all posts
GET https://jsonplaceholder.typicode.com/posts

Click “Send Request” above it.

You’ll see the response in a new VS Code tab.

That’s it. No GUI required.

Example 2 — POST Request

### Create new post
POST https://jsonplaceholder.typicode.com/posts
Content-Type: application/json

{
  "title": "API Testing in VS Code",
  "body": "Using REST Client instead of Postman",
  "userId": 1
}

Clean. Readable. Commit-ready.

4️⃣ Separating Multiple Requests (CRITICAL)

When you have multiple requests in one file, you must separate them using:

###

Correct:

### Get posts
GET https://jsonplaceholder.typicode.com/posts

###

### Get single post
GET https://jsonplaceholder.typicode.com/posts/1

🚨 BIG ISSUE #1 — `---` vs `###`

Many developers coming from Markdown or other tools use:

---

This is WRONG for REST Client.

If you write:

POST https://example.com/api
Content-Type: application/json

{
  "name": "Ajit"
}
---

VS Code will send this entire block:

{
  "name": "Ajit"
}
---

And your backend (for example, Django REST Framework) will throw:

JSON parse error - Extra data

Because:

--- is treated as part of the JSON body.

✅ Rule:

Only ### separates requests.

Nothing else works.

5️⃣ Using Variables (Cleaner & Reusable)

Instead of hardcoding URLs:

@baseUrl = https://jsonplaceholder.typicode.com

### Get posts
GET {{baseUrl}}/posts

Works immediately.

But this doesn’t solve environment-specific values like tokens.

6️⃣ Setting Up Environments (Local & Production)

There are two approaches:

❌ Attempt 1 — `rest-client.environment.json` (Did Not Work Initially)

You create:

rest-client.environment.json

In project root:

{
  "local": {
    "baseUrl": "http://127.0.0.1:8000",
    "authToken": "real-token"
  },
  "prod": {
    "baseUrl": "https://api.example.com",
    "authToken": "prod-token"
  }
}

And use:

GET {{baseUrl}}/api/users
Authorization: Token {{authToken}}

Expected behavior:
Bottom-right environment selector appears.

Problem:
It didn’t appear.

Why?

Because in some workspace setups:

Multi-root workspace
Custom VS Code config
REST Client not auto-detecting root file

The extension didn’t auto-load the file.

✅ Final Working Solution — `.vscode/settings.json`

Inside project root:

.vscode/settings.json

Add:

{
  "rest-client.environmentVariables": {
    "local": {
      "baseUrl": "http://127.0.0.1:8000",
      "authToken": "your-real-token"
    },
    "prod": {
      "baseUrl": "https://api.example.com",
      "authToken": "prod-token"
    }
  }
}

Reload VS Code.

Now:

Open .http
Bottom-right shows environment selector
Choose local or prod

Variables resolve correctly.

Problem solved.

⚠️ Important Security Note

.vscode/settings.json now contains tokens.

That means:

You MUST add this to `.gitignore`:

.vscode/

Otherwise you risk committing:

API tokens
Secrets
Environment configs

Safer production pattern:

.vscode/settings.json          (ignored)
.vscode/settings.example.json  (committed template)

Example template:

{
  "rest-client.environmentVariables": {
    "local": {
      "baseUrl": "http://127.0.0.1:8000",
      "authToken": "paste-token-here"
    }
  }
}

7️⃣ Final Clean Project Structure

project/
│
├── api.http
├── .vscode/
│   └── settings.json   (ignored)
├── .gitignore
└── README.md

.gitignore:

.vscode/

8️⃣ When Should You Still Use Postman?

Use Postman when:

You need automated test scripting
Team collaboration on collections
Complex OAuth flows
Mock servers

Use api.http when:

You want repo-based documentation
Backend development speed matters
You want zero tool switching
You prefer Git-tracked API specs

9️⃣ Final Thoughts

For experienced developers, api.http with REST Client feels like:

curl, but readable
Postman, but version-controlled
Documentation, but executable

It removes friction.

And once you fix:

--- vs ###
Environment configuration quirks

It becomes a powerful, lightweight part of your backend workflow.

Stop Clicking Around: API Testing with `.http` Files (From Beginner to Pro)

Ajit Kumar — Mon, 16 Feb 2026 10:12:24 +0000

Modern API development does not require heavyweight GUI tools. If you are building APIs with Django REST Framework or any backend stack, you can test, document, and version your APIs using simple .http files — directly inside Visual Studio Code.

This tutorial walks from beginner to advanced usage of the .http approach, compares it with tools like Postman, HTTPie, and Swagger, and shows how to build a clean, executable API workflow.

Why `.http` Files?

A .http file is a plain text file containing HTTP requests.

Example:

GET http://127.0.0.1:8000/api/blogs/
Accept: application/json

With the REST Client extension in VS Code, you can:

Click “Send Request”
See status code
View headers
Inspect JSON response
Measure response time

All without leaving your editor.

This gives you:

Executable documentation
Version-controlled API tests
Zero external dependency
Simple text-based workflow

Part 1 — Beginner Level

Step 1: Install REST Client

Open VS Code → Extensions → Install:

REST Client (by Huachao Mao)

Step 2: Create `api.http`

Create a file:

api.http

Add:

@baseUrl = http://127.0.0.1:8000

### List Blogs
GET {{baseUrl}}/api/blogs/
Accept: application/json

Start your server:

python manage.py runserver

Click Send Request.

You’ll see:

200 OK
Response body
Headers
Timing

That’s it.

You just tested your API with a text file.

Part 2 — Intermediate Usage

Now let’s add CRUD operations.

### Create Blog
POST {{baseUrl}}/api/blogs/
Content-Type: application/json

{
  "title": "API Testing Simplified",
  "content": "Using .http files improves workflow."
}

### Get Blog by ID
GET {{baseUrl}}/api/blogs/1/

### Update Blog
PUT {{baseUrl}}/api/blogs/1/
Content-Type: application/json

{
  "title": "Updated Title",
  "content": "Updated content."
}

### Delete Blog
DELETE {{baseUrl}}/api/blogs/1/

Now your file:

Documents endpoints
Stores example payloads
Acts as test suite
Becomes team reference

Part 3 — Advanced Techniques

1️⃣ Environment Switching

At the top:

@baseUrl = http://127.0.0.1:8000

Change to:

@baseUrl = https://staging.myapp.com

All requests update automatically.

2️⃣ Variables from Response

You can extract response data:

### Create Blog
POST {{baseUrl}}/api/blogs/
Content-Type: application/json

{
  "title": "Dynamic Test",
  "content": "Testing extraction"
}

Then reference:

GET {{baseUrl}}/api/blogs/{{createBlog.response.body.id}}/

Now requests become chained.

3️⃣ Authentication Headers

@token = your_jwt_token_here

GET {{baseUrl}}/api/secure-endpoint/
Authorization: Bearer {{token}}

4️⃣ Expected Response Documentation

You can document expected behavior:

# Expected 201 Created
# Metadata should auto-generate

POST {{baseUrl}}/api/blogs/
Content-Type: application/json

{
  "title": "AI",
  "content": "AI is powerful."
}

This becomes living API documentation.

How It Compares to Other Tools

1️⃣ Postman

Postman is powerful:

Pros:

GUI
Team collaboration
Automated test scripts
Cloud sync

Cons:

Requires account
Harder to version-control cleanly
Collections stored outside codebase
Heavy UI

.http advantage:

Lives in repo
Lightweight
No login
Git diff-friendly

2️⃣ HTTPie

HTTPie is CLI-based.

Example:

http GET http://127.0.0.1:8000/api/blogs/

Pros:

Beautiful CLI output
Fast
Scriptable

Cons:

Not self-documenting
Not clickable
Harder for beginners

.http sits between curl and Postman — readable and interactive.

3️⃣ Swagger / OpenAPI

Swagger auto-generates API documentation.

Pros:

Interactive UI
Schema validation
Industry standard

Cons:

Generated, not curated
Harder to add narrative examples
Not ideal for manual test workflows

Best practice:

Use Swagger for:

Public documentation
API contracts

Use .http for:

Development testing
Internal API examples
Regression testing

Why `.http` is Excellent for Teams

Stored in Git
Reviewable via pull request
Encourages documented APIs
Works offline
Zero SaaS dependency
Reproducible environments

Your repository now contains:

api.http

Which acts as:

API spec
Test runner
Example library
Debug tool

Scaling the Pattern

For large systems:

/api-tests/
    blog.http
    auth.http
    analytics.http

Organize per domain.

For multi-version APIs:

/api-tests/v1/

When Not to Use `.http`

Load testing (use k6 or Locust)
Automated CI assertions (use pytest + requests)
Public API documentation portal (use OpenAPI)

It is a developer-first tool, not an enterprise API gateway.

Final Recommendation

For solo developers, small teams, backend-focused engineers:

.http + VS Code is the fastest, cleanest API testing workflow available.

It removes friction.
It keeps documentation close to code.
It scales from beginner to advanced usage.

And it avoids unnecessary tooling overhead.

The Complete Guide to Ollama: Run Large Language Models Locally

Ajit Kumar — Mon, 16 Feb 2026 09:19:48 +0000

Imagine having the power of GPT-4 or Claude running entirely on your laptop—no internet required, no API costs, and complete privacy. Until recently, this seemed like a distant dream reserved for tech giants with massive server farms. But the landscape has changed dramatically. Thanks to Ollama, anyone with a modern computer can now run sophisticated AI models locally, whether you're coding on a plane at 35,000 feet, analyzing sensitive documents that can never touch the cloud, or simply experimenting with AI without watching your API bill climb. This comprehensive guide will walk you through everything you need to know about Ollama—from your first installation to building production-ready AI-powered applications that respect your privacy and your wallet.

Part A: Setting Up Ollama

Introduction to Ollama

Ollama is a powerful, open-source tool that enables you to run large language models (LLMs) locally on your own machine. Think of it as Docker for AI models—it packages everything you need to run models like Llama, Mistral, CodeLlama, and dozens of others into easy-to-use containers that work seamlessly on your computer.

Why use Ollama?

Privacy: Your data never leaves your machine. Perfect for sensitive code, personal documents, or proprietary information.
No API costs: Run unlimited queries without paying per token.
Offline capability: Work without internet connectivity once models are downloaded.
Speed: No network latency—responses are generated locally.
Experimentation: Try different models and configurations freely.
Integration: Works with various tools and development environments.

Installation

Ollama supports macOS, Linux, and Windows. The installation process is straightforward for all platforms.

macOS

Download the official installer from ollama.com:

# Or use Homebrew
brew install ollama

Linux

Run the installation script:

curl -fsSL https://ollama.com/install.sh | sh

This script supports most Linux distributions including Ubuntu, Debian, Fedora, and CentOS.

Windows

Download the Windows installer from ollama.com or use the preview version. Ollama on Windows runs natively and supports the same commands as other platforms.

Docker Installation

For containerized deployment:

docker pull ollama/ollama
docker run -d -v ollama:/root/.ollama -p 11434:11434 --name ollama ollama/ollama

Testing Your Installation

After installation, verify Ollama is working:

# Check version
ollama --version

# Check if the service is running
ollama serve

You should see output indicating the Ollama server is running on http://localhost:11434.

To test with a simple model:

# Pull and run a small model
ollama run llama3.2

# You should get an interactive prompt
# Try: "Hello, how are you?"

If you see a response from the model, congratulations—Ollama is working correctly!

Updating Ollama

Keeping Ollama updated ensures you have the latest features, bug fixes, and security patches.

Why Update?

New model support: Access to the latest model architectures and versions
Performance improvements: Faster inference and better memory management
Bug fixes: Resolved issues and stability improvements
New features: Enhanced CLI commands and API capabilities
Security patches: Important security updates

How to Update

macOS (Homebrew):

brew update
brew upgrade ollama

macOS (Installer):
Download and run the latest installer from ollama.com—it will update your existing installation.

Linux:

curl -fsSL https://ollama.com/install.sh | sh

The script automatically detects and updates your existing installation.

Windows:
Download and run the latest installer.

Docker:

docker pull ollama/ollama:latest
docker stop ollama
docker rm ollama
docker run -d -v ollama:/root/.ollama -p 11434:11434 --name ollama ollama/ollama

After updating, verify the new version:

ollama --version

Understanding Ollama Models

Ollama supports a vast ecosystem of models, each optimized for different tasks. Here's a breakdown of the major categories:

General Purpose Models

Llama 3.2 (1B, 3B) - Meta's latest efficient models

Best for: General conversation, reasoning, lightweight tasks
Size: 1B parameters (740MB), 3B parameters (2GB)
Command: ollama run llama3.2 or ollama run llama3.2:1b

Llama 3.1 (8B, 70B, 405B) - Meta's powerful flagship models

Best for: Complex reasoning, long context (128K tokens)
Size: 8B (4.7GB), 70B (40GB), 405B (231GB)
Command: ollama run llama3.1 or ollama run llama3.1:70b

Mistral (7B) - Efficient and capable model

Best for: General purpose tasks with good speed/quality balance
Size: 4.1GB
Command: ollama run mistral

Mixtral (8x7B, 8x22B) - Mixture of Experts models

Best for: High-quality outputs with efficient inference
Size: 26GB (8x7B), 80GB (8x22B)
Command: ollama run mixtral

Coding Models

CodeLlama (7B, 13B, 34B, 70B) - Specialized for code

Best for: Code generation, completion, debugging
Size: 7B (3.8GB), 13B (7.4GB), 34B (19GB), 70B (39GB)
Command: ollama run codellama

DeepSeek Coder (6.7B, 33B) - Strong coding performance

Best for: Multiple programming languages, code explanation
Size: 6.7B (3.8GB), 33B (19GB)
Command: ollama run deepseek-coder

Qwen 2.5 Coder (1.5B, 7B, 32B) - Alibaba's coding model

Best for: Code generation with multilingual support
Command: ollama run qwen2.5-coder

Specialized Models

Llama 3.2 Vision (11B, 90B) - Multimodal understanding

Best for: Image analysis, visual question answering
Command: ollama run llama3.2-vision

Phi-3 (3.8B, 14B) - Microsoft's small language models

Best for: Efficient deployment, edge devices
Size: 3.8B (2.3GB), 14B (7.9GB)
Command: ollama run phi3

Gemma 2 (2B, 9B, 27B) - Google's open models

Best for: Safety-focused applications
Command: ollama run gemma2

Neural Chat (7B) - Fine-tuned for conversations

Best for: Interactive dialogue
Command: ollama run neural-chat

How to Choose a Model

For general use on consumer hardware: llama3.2:3b or mistral
For coding tasks: codellama:7b or deepseek-coder:6.7b
For best quality (with powerful GPU): llama3.1:70b or mixtral:8x22b
For resource-constrained systems: phi3:3.8b or llama3.2:1b
For vision tasks: llama3.2-vision

Starting and Stopping Ollama

Starting Ollama

The Ollama server needs to be running to handle model requests.

On macOS and Linux:

# Start in foreground (see logs)
ollama serve

# Or run as background service (automatically starts on boot)
# On macOS, Ollama runs as a LaunchAgent
# On Linux with systemd:
systemctl start ollama

On Windows:
Ollama runs as a service automatically after installation. Access it via the system tray icon.

Verify it's running:

curl http://localhost:11434
# Should return: "Ollama is running"

Stopping Ollama

On macOS:

# If running in foreground, press Ctrl+C
# If running as service:
launchctl unload ~/Library/LaunchAgents/com.ollama.ollama.plist

On Linux:

# If running in foreground, press Ctrl+C
# If running as systemd service:
systemctl stop ollama

On Windows:
Right-click the Ollama icon in the system tray and select "Quit Ollama."

Docker:

docker stop ollama

Serving a Specific Model

Once Ollama is running, you can interact with models in several ways.

Interactive Mode

Run a model and chat directly in the terminal:

ollama run llama3.2

This starts an interactive session. Type your prompts and press Enter. Type /bye to exit.

Pull a Model Without Running

Download a model for later use:

ollama pull codellama:13b

List Downloaded Models

See which models you have locally:

ollama list

Output shows model name, size, and last modified date.

Run with Custom Parameters

Control model behavior with parameters:

ollama run llama3.2 --verbose

# Or set parameters in the prompt
ollama run llama3.2 "Write a haiku" --temperature 0.8 --top-p 0.9

API Mode (Serving to Other Applications)

Ollama exposes an OpenAI-compatible API on http://localhost:11434:

# Generate a completion
curl http://localhost:11434/api/generate -d '{
  "model": "llama3.2",
  "prompt": "Why is the sky blue?",
  "stream": false
}'

# Chat completion
curl http://localhost:11434/api/chat -d '{
  "model": "llama3.2",
  "messages": [
    {"role": "user", "content": "Hello!"}
  ]
}'

This API allows integration with countless tools and applications.

Remove Models You Don't Need

Free up disk space:

ollama rm codellama:70b

Quick Reference Commands

# Installation & Updates
ollama --version              # Check version
curl -fsSL https://ollama.com/install.sh | sh  # Install/update

# Model Management
ollama pull <model>           # Download a model
ollama list                   # List local models
ollama rm <model>             # Delete a model
ollama show <model>           # Show model information

# Running Models
ollama run <model>            # Interactive mode
ollama run <model> "prompt"   # One-off generation
ollama serve                  # Start API server

# API Testing
curl http://localhost:11434   # Check if running
curl http://localhost:11434/api/tags  # List models via API

Part B: Real-World Use Cases

Now that you have Ollama set up, let's explore powerful ways to use it in your daily workflow.

Use Case 1: Local Coding with Claude Code and AI IDEs

One of the most powerful applications of Ollama is integrating local models with development tools. This gives you AI-powered coding assistance that's private, fast, and free.

Using Ollama with Claude Code

Claude Code is a command-line agentic coding tool. While it typically uses Anthropic's API, you can configure it to work with local models via Ollama for certain tasks.

Setup:

Install Claude Code:

npm install -g @anthropic-ai/claude-code

Configure to use Ollama as a secondary tool for code completion:

# You can use Ollama's OpenAI-compatible endpoint
export OLLAMA_API_BASE=http://localhost:11434/v1

Workflow:

Use Claude Code with Anthropic's API for complex agentic tasks (planning, multi-file refactoring)
Use local Ollama models for quick completions, code explanations, or offline work
Combine both: Claude for architecture, Ollama for implementation details

Using Ollama with Cursor

Cursor is a popular AI-powered IDE. Configure it to use Ollama:

Open Cursor Settings
Navigate to Models
Add custom model endpoint: http://localhost:11434/v1
Select your preferred model (e.g., codellama:13b)

Benefits:

No API costs for code completions
Faster responses (no network latency)
Private code never leaves your machine

Using Ollama with Continue.dev

Continue is a VS Code / JetBrains extension that brings AI assistance to your IDE:

Install the Continue extension
Configure ~/.continue/config.json:

{
  "models": [
    {
      "title": "CodeLlama Local",
      "provider": "ollama",
      "model": "codellama:13b",
      "apiBase": "http://localhost:11434"
    }
  ]
}

Use Cmd/Ctrl+L to chat with your local model
Highlight code and ask questions, generate tests, or refactor

Using Ollama with Tabby (Self-Hosted Code Completion)

Tabby provides GitHub Copilot-like code completion:

# Run Tabby with Ollama backend
docker run -it --gpus all \
  -p 8080:8080 \
  -v ~/.tabby:/data \
  tabbyml/tabby \
  serve --model TabbyML/CodeLlama-7B --device cuda

Configure your IDE to point to http://localhost:8080.

Best Practices for Local Coding Models

Choose the right model size: 7B models for real-time completions, 13B+ for deeper analysis
Use code-specific models: CodeLlama, DeepSeek Coder, or Qwen Coder perform better than general models
Combine tools: Use local models for most tasks, cloud models for complex reasoning
Fine-tune prompts: Local models benefit from clear, specific instructions
Monitor resource usage: Large models need significant RAM/VRAM

Use Case 2: Document Ingestion and RAG Workflows

Retrieval-Augmented Generation (RAG) is a technique where you feed your own documents to an LLM, enabling it to answer questions based on your specific data. Ollama is perfect for building private RAG systems.

What is RAG?

Traditional LLMs only know what they were trained on. RAG extends their knowledge by:

Ingesting your documents (PDFs, text files, web pages)
Chunking them into smaller pieces
Embedding these chunks into vector representations
Storing embeddings in a vector database
Retrieving relevant chunks when you ask a question
Generating answers using both the retrieved context and the LLM

Simple RAG with Ollama and Python

Here's a basic RAG implementation:

# Install dependencies
# pip install ollama chromadb langchain pypdf --break-system-packages

import ollama
import chromadb
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader

# 1. Load and chunk documents
loader = PyPDFLoader("your_document.pdf")
documents = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)

# 2. Create embeddings and store in vector DB
client = chromadb.Client()
collection = client.create_collection("my_docs")

for i, chunk in enumerate(chunks):
    # Use Ollama to generate embeddings
    embedding = ollama.embeddings(
        model="nomic-embed-text",
        prompt=chunk.page_content
    )["embedding"]

    collection.add(
        ids=[str(i)],
        embeddings=[embedding],
        documents=[chunk.page_content]
    )

# 3. Query the system
def ask_question(question):
    # Get embedding for question
    question_embedding = ollama.embeddings(
        model="nomic-embed-text",
        prompt=question
    )["embedding"]

    # Find relevant chunks
    results = collection.query(
        query_embeddings=[question_embedding],
        n_results=3
    )

    # Build context from retrieved chunks
    context = "\n\n".join(results["documents"][0])

    # Generate answer using Ollama
    response = ollama.chat(
        model="llama3.2",
        messages=[{
            "role": "user",
            "content": f"Context:\n{context}\n\nQuestion: {question}"
        }]
    )

    return response["message"]["content"]

# Use it
answer = ask_question("What are the main points in the document?")
print(answer)

Advanced RAG with Ollama and Open Source Tools

Using Ollama with LangChain:

from langchain.llms import Ollama
from langchain.chains import RetrievalQA
from langchain.vectorstores import FAISS
from langchain.embeddings import OllamaEmbeddings

# Initialize Ollama
llm = Ollama(model="llama3.2")
embeddings = OllamaEmbeddings(model="nomic-embed-text")

# Create vector store
vectorstore = FAISS.from_documents(chunks, embeddings)

# Create QA chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    chain_type="stuff"
)

# Ask questions
result = qa_chain.run("What does the document say about X?")

Using Ollama with AnythingLLM:

AnythingLLM is a full-stack RAG application with a beautiful UI:

Install: Download from anythingllm.com
Configure Ollama as LLM provider in settings
Upload documents via the UI
Start chatting with your documents

Using Ollama with Quivr (Your Second Brain):

Quivr is a RAG application for personal knowledge management:

# Clone and setup
git clone https://github.com/QuivrHQ/quivr
cd quivr
docker-compose up

# Configure to use Ollama in .env file
LLM_PROVIDER=ollama
OLLAMA_API_BASE=http://localhost:11434

Ingestion Workflow Best Practices

1. Document Preprocessing:

Clean text (remove headers, footers, page numbers)
Extract tables and images separately if needed
Use OCR for scanned documents (tesseract-ocr)

2. Chunking Strategies:

Fixed size: Simple, works for uniform content (500-1000 tokens)
Semantic: Split by paragraphs or sections (better context)
Recursive: LangChain's RecursiveCharacterTextSplitter (balanced)

3. Embedding Models:

nomic-embed-text: Best all-around, 137M parameters
mxbai-embed-large: Highest quality, 335M parameters
all-minilm: Fastest, lightweight option

4. Retrieval Optimization:

Use hybrid search (keyword + semantic)
Re-rank results with a cross-encoder
Adjust number of retrieved chunks (typically 3-5)

5. Prompt Engineering:

Context: [retrieved chunks]

Based ONLY on the context above, answer the following question.
If the answer is not in the context, say "I don't have enough information."

Question: {user_question}

Answer:

Production RAG Considerations

Vector Database: Use Qdrant, Weaviate, or Pinecone for scale
Caching: Cache embeddings to avoid recomputation
Monitoring: Track retrieval accuracy and response quality
Updates: Implement incremental updates when documents change
Multi-modal: Use vision models for PDFs with images/charts

Performance Tips for Ollama

GPU Acceleration: Ollama automatically uses GPU if available (NVIDIA, AMD, or Apple Silicon)
Memory Management: Larger models need more RAM/VRAM (check with ollama show <model>)
Concurrent Requests: Ollama handles multiple requests, but memory is shared
Model Quantization: Use quantized models (Q4, Q5) for better performance with acceptable quality loss

Conclusion

Ollama opens up a world of possibilities for running powerful AI models on your own hardware. From coding assistance to document analysis, from privacy-preserving workflows to cost-free experimentation, the applications are endless.

Key Takeaways:

Ollama makes running local LLMs accessible and practical
Choose models based on your hardware and use case
Integration with development tools enhances productivity
RAG workflows unlock the power of your private data
Local models provide privacy, speed, and zero API costs

Next Steps:

Install Ollama and test a few models
Integrate with your favorite IDE or coding tool
Build a simple RAG system with your documents
Explore the Ollama model library at ollama.com/library
Join the community at github.com/ollama/ollama

Happy local LLM exploration! 🚀

Why Your Deep Learning Job Dies After SSH Logout — A Practical Guide to Persistent Linux Sessions

Ajit Kumar — Mon, 16 Feb 2026 05:28:43 +0000

You SSH into a remote server.
You start a long-running deep learning training job:

python train.py

Everything works. GPU utilization is high. Logs are printing.

Then your laptop sleeps.
Or WiFi drops.
Or you close the terminal.

You reconnect to the server…

Your training job is gone.

But on another server, your training continues even after you disconnect.

Why?

This is not random behavior. It is deterministic Unix session management.

Let’s break it down precisely.

The Root Cause: Controlling Terminals and `SIGHUP`

When you SSH into a Linux server:

SSH creates a login session.
A shell process (e.g., bash) is started.
That shell is attached to a controlling terminal (TTY).
Any command you run becomes a child process of that shell.

Process tree example:

sshd
 └── bash
      └── python train.py

When your SSH session ends:

The terminal disappears.
The shell receives a SIGHUP (Hang Up signal).
The shell forwards SIGHUP to its child processes.
Your Python training process exits.

This is default POSIX behavior.

The operating system is doing exactly what it is designed to do.

Why It Works on Another Server

If training survives logout on another machine, one of these is true:

It was started inside tmux
It was started inside screen
It was launched using nohup
It was disowned
It was started as a systemd service
The SSH daemon is configured differently (rare but possible)

The most common reason: tmux is being used.

Real-World Scenario (Deep Learning Context)

Imagine this:

You launch a 3-day transformer training run.
You’re using expensive GPU hardware.
Model checkpoints save every 2 hours.
Your internet drops after 4 hours.

Without proper session handling:

Training dies immediately.
GPU memory is released.
You lose progress since the last checkpoint.
Compute time is wasted.
Experiment reproducibility is impacted.

In production ML workflows, this is unacceptable.

Proper Solutions (Ranked by Maturity)

1️⃣ tmux — The Research Standard

tmux creates a persistent terminal multiplexer. Your training process attaches to tmux — not directly to SSH.

Install:

sudo apt install tmux

Start session:

tmux new -s training

Run training:

python train.py

Detach safely:

Ctrl + B, then D

Logout freely. Training continues.

Reconnect later:

tmux attach -t training

Why this works:

tmux owns the controlling TTY.
SSH disconnection does not kill tmux.
Your process remains alive.

This is the recommended workflow for ML engineers and researchers.

2️⃣ nohup — Minimal Fix

nohup python train.py > train.log 2>&1 &

What it does:

Ignores SIGHUP
Redirects output
Runs in background

Good for:

Simple batch jobs
Fire-and-forget scripts

Downside:

No interactive recovery
Harder process management

3️⃣ disown — Shell-Level Detachment

python train.py &
disown

This removes the process from the shell job table.

Less robust than tmux but works.

4️⃣ systemd — Production-Grade

For structured environments:

Create a service file:

/etc/systemd/system/ml-training.service

Then:

sudo systemctl daemon-reload
sudo systemctl start ml-training

Benefits:

Auto restart on failure
Resource control
Logging via journalctl
Reproducible startup
Can run at boot

This is appropriate for:

Production ML pipelines
Enterprise GPU clusters
Persistent inference services

How to Diagnose Your Current Setup

Before disconnecting, check:

ps -o pid,ppid,cmd -p <python_pid>

If the parent process (PPID) is your shell → it will die when SSH disconnects.

Check if you are inside tmux:

echo $TMUX

If empty → you are not using tmux.

Server Admin Configuration Considerations

From the system administration side, there are several configurations that impact user sessions.

1️⃣ SSH Daemon Settings

File:

/etc/ssh/sshd_config

Relevant options:

ClientAliveInterval
ClientAliveCountMax
TCPKeepAlive

These control idle timeout behavior — but they do NOT prevent SIGHUP when a session closes.

They only affect when the connection is dropped.

2️⃣ systemd User Session Settings

Modern Linux distributions use systemd-logind.

File:

/etc/systemd/logind.conf

Key option:

KillUserProcesses=yes

If enabled:

All user processes are killed when the session ends.
Even tmux sessions may die.

To allow persistent processes:

KillUserProcesses=no

After change:

sudo systemctl restart systemd-logind

This setting is critical on shared GPU servers.

3️⃣ Resource Policies

Admins may enforce:

cgroups limits
SLURM job scheduler policies
Idle job cleanup scripts

On managed clusters, processes outside scheduler control may be automatically terminated.

In HPC environments, users should launch jobs via:

sbatch
srun
qsub
Kubernetes job manifests

Not direct SSH shells.

When Admin Intervention Is Required

If:

tmux sessions still die on logout
background processes terminate unexpectedly
training stops even with nohup

Then the admin should verify:

KillUserProcesses setting
PAM session configuration
Custom logout hooks
Cluster job scheduler enforcement
cgroup cleanup policies

On well-configured ML servers, tmux should work without special privileges.

Best Practice for ML Engineers

For experimentation:

Always use tmux
Enable periodic checkpointing
Log to file
Monitor GPU usage via nvidia-smi

For production:

Use systemd or a scheduler
Configure restart policies
Use structured logging
Implement health checks

Final Takeaway

Your training stops because:

It is attached to a shell that receives SIGHUP when SSH disconnects.

This is expected behavior.

The solution is not “keep SSH open.”

The solution is:

Decouple your process from the SSH session.
Use tmux, nohup, or systemd appropriately.
Ensure systemd-logind is not killing user processes.

Once configured correctly, your deep learning jobs should survive network instability, laptop sleep, and terminal closure — exactly as they should in a robust ML workflow.

DEV Community: Ajit Kumar

Moving a production API to Hetzner, part 3: migrating data and cutting over

0. The example application

1. Decide what moves, and what doesn't

2. Copying between two servers through the laptop

The trick

Verify it arrived intact

3. Pull the image, never build on the box

4. Start the data services only

5. Rehearse the restore before you need it

A bug a rehearsal like this can turn up

6. DNS: lower the TTL a day ahead

How DNS caching works

DNS-only, not proxied

7. The cutover

Verification checklist

8. Rollback, and the one rule

9. Decommission, slowly

What the whole series boils down to

Moving a production API to Hetzner, part 2: hardening and configuring the server

1. ~/.ssh/config: two servers, two keys, short commands

The problem it solves

Why IdentitiesOnly yes

2. First login, and the plan for not locking yourself out

The lockout problem

3. Pass 1, explained block by block

set -euo pipefail

The user

Packages

sshd: drop-in files and "first value wins"

Why ssh.socket and not just ssh

ufw: the host firewall

Swap

4. Test, then pass 2

5. Verify like an attacker

Automatic updates, and the reboot they don't do

6. Reading a port test: open, refused, filtered

7. Docker

Where we are

Moving a production API to Hetzner, part 1: choosing and creating the server

What we are moving

1. Picking a server type

The three families

How big does it have to be?

"Limited availability" is real

Location and image

2. SSH keys: how you'll get in

What a key pair is

Creating the key

3. Primary IPs: an address that outlives the server

What a Primary IP is

Why "auto delete" matters

4. The Cloud Firewall: the perimeter outside your VM

Two places a firewall can sit

Stateful filtering, in one paragraph

The rules

5. cloud-init: configuring the server before you first log in

What it is

The order cloud-init works in

What actually happened

6. Create

What we have so far

One .env file, three answers: measuring how dotenv readers disagree

Measured environment

1. The line that started this

2. A tiny harness

Fixtures

3. Readers and adapters

Python: python-dotenv

Node: dotenv, with and without dotenv-expand

Bash: source

Docker: Compose and docker run --env-file

4. Comparing the readers automatically

Regenerating everything

Extending the harness

5. Results

6. Observations

What this means

7. Why this matters in practice

8. Bonus: differential fuzzing

1. `~/.ssh/config`: two servers, two keys, short commands

Why `IdentitiesOnly yes`

`set -euo pipefail`

Why `ssh.socket` and not just `ssh`

Bash: `source`

Docker: Compose and `docker run --env-file`

Part 2: Years of `.env` Pain Across Projects

Part 4: From `dotenv-cli` to `evnx` — A Name Born From Distribution