Lesson 05 of 05 · published

Build Your Own Channel

~15 min · capstone, http-server, resumable-transfer, sha256, stdlib, checklist

Level 0Loose File

0 XP0/43 lessons0/12 achievements

0/100 XP to next level100 XP to go0% complete

Small Enough to Read in One Sitting

Delta is over nine thousand lines of Python because it has to survive a family's real network: three kinds of route, tickets, pinned TLS, a disk cache, parallel workers, bundles, stall timeouts, links, pairs and a write call. But the rules that make it trustworthy fit in a page. This lesson is that page as running code: minichannel.py, one file of standard-library Python that serves a folder on loopback and reads it back, and keeps seven of the quest's promises. Run it with no arguments and it builds a temporary share, starts its owner on a free port, and drives every rule against it: python3 minichannel.py.

It runs on Apple's /usr/bin/python3 (3.9) as well as anything newer, which is the same constraint Delta lives under.

Seven Promises, and Where Each Lives

Promise	In `minichannel.py`	In Delta
Refusals are about places	`clean_rel` refuses `..` and NUL, and any path with a component in `REFUSED`, before anything is opened	`clean_rel`, `CREDENTIAL_*`, the TCC guard, checked as asked and with links followed (Track 2)
Loopback is not identity	every request must carry the token, compared with `secrets.compare_digest`	`X-Delta-Local-Token`, a 0600 file only the account can read (Track 4)
Validators come from the owner	`validator(st)` is size, mtime in nanoseconds and inode; `If-None-Match` gets a 304	the owner's ETag, or a child's own source tag; never a TTL (Track 5)
Every error has a named cause	`ChannelError(cause, detail, status)` crosses the wire as a JSON document with `error`, `detail` and `status`, and is rebuilt on the reader's side	`delta/errors.py`, every failure naming what, where and why (Track 1)
A short body is an error	the reader catches the `http.client.IncompleteRead` a body that ends early raises, keeps what did arrive, and compares it with `Content-Length`: a cut comes back as a named `unavailable` saying how far it got	`peer.Response`, because `read(n)` returns short without a word (Track 3)
Verified before recorded	each chunk is checked against the source's SHA-256, written at its offset, fsynced, and only then appended to the record	`Partial`: one appended line per chunk (Track 6)
Lands whole or not at all	the file's chunk list must equal the source's, then `os.replace` moves it into place	the atomic rename after the chunk list matches (Track 6)

Two smaller rules ride along. The owner reads (drains) any request body before it answers, even with a refusal, so a refusal never strands bytes in a kept-alive connection (Track 6). And the partial's name includes the source's validator, so if the file changes between runs the next pull starts a fresh partial instead of splicing two versions together. The old one is simply abandoned, which is why Delta sweeps partials untouched for fourteen days (lesson 2 of this track).

Walking the Demo

The run prints six numbered results (the second code block shows them). Each one is a promise kept, and you should be able to say which line of code kept it:

The first read is a miss: bytes crossed. The second sends the validator it holds and gets a 304: revalidated, no bytes.
The first pull is cut on purpose after two chunks. The cut is an error with a cause, not a short file.
The second pull reads its record, says how many chunks are already verified, asks only for the rest, checks the whole chunk list and renames the file into place. The demo then compares the SHA-256 of source and copy.
A .. path, a path into a refused place and a missing file each come back as a named cause: forbidden, forbidden, not_found.
A wrong token is unauthorized, before the path is even looked at.
A second owner, Cutter, promises the whole file in its Content-Length and closes the connection halfway. The reader does not hand back half a file: it raises unavailable, "closed mid-body at N of M". Two experiments show what each line buys. Remove the inner except http.client.IncompleteRead and the cut is still named, but by the outer handler, which can only repeat the exception's text. Then narrow the outer handler to except OSError and the cut escapes as an unnamed traceback: IncompleteRead is an http.client.HTTPException, not an OSError.

What It Leaves Out, on Purpose

Everything below is real in Delta and absent here. Each is a good next step, roughly in the order the quest taught them:

Routes and identity between two real Macs. Here both ends are one process on loopback. Across Macs you need Track 3's on-link choice and Track 4's roster, tickets and pinned certificates.
A cache that survives a restart. The reader's cache is a dict. Delta keeps blobs on disk named by SHA-256, bounded by size and by the disk floor, and labels a copy stale when the owner cannot be reached.
Parallel workers, bundles and stall timeouts. Chunks here move one at a time and each request has one 30-second timeout. Delta spreads 16 MiB chunks over several workers, batches small files, and times out only on a stall.
Jobs. A pull here lives as long as your terminal. Delta runs it inside the daemon and lets you follow it.
Two-way. A named pair with an archive, and a write call sent at most once, are Track 7.

The Checklist You Take Home

Whatever you build that moves data between machines, these are the questions the quest leaves you with:

Is data riding a channel you own and can read the code of, while commands keep their own?
Which single process may dial the network, and do the others ask it locally?
Is every refusal a rule about a place, checked before the place is touched?
Is the route chosen from facts you verified, never a name?
Is identity proven on a channel that already authenticates, and is anything you hand out stored only as a hash?
Does every cached answer come with the owner's validator, and does a stale one say so?
Is every chunk verified before it is recorded, and does a re-run resume?
Does a job end done only when every planned item landed or is named?
Is a write that must not happen twice sent at most once, with did it leave? as a question you can answer?
Does everything that accumulates have a number, and is that number in the health document?

Code

minichannel.py: the quest's rules in one runnable file·python

# minichannel.py: a two-endpoint data channel in the standard library only.
# Run it with no arguments: it serves a temporary share on loopback and drives
# every rule the quest taught against it. Python 3.9+, macOS or Linux.
import hashlib, http.client, json, os, secrets, tempfile, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import quote, unquote, urlsplit

CHUNK = 1024 * 1024                       # Delta uses 16 MiB; small here so a demo has chunks
REFUSED = {".ssh", ".gnupg", ".env"}      # places, not files: refused at any component


class ChannelError(Exception):
    """A named cause: what failed, in a sentence, with its HTTP status."""
    def __init__(self, cause, detail, status):
        super().__init__(f"{cause}: {detail}")
        self.cause, self.detail, self.status = cause, detail, status


def clean_rel(rel):
    parts = [p for p in unquote(rel).split("/") if p not in ("", ".")]
    if "\x00" in rel or ".." in parts:
        raise ChannelError("forbidden", "'..' and NUL are not path components", 403)
    if REFUSED & set(parts):
        raise ChannelError("forbidden", f"{rel} is inside a refused place", 403)
    return "/".join(parts)


def validator(st):
    """The owner's word for a version: changes when the bytes can have changed."""
    return f'"{st.st_size:x}-{st.st_mtime_ns:x}-{st.st_ino:x}"'


def make_handler(root, token):
    class Owner(BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"     # keep-alive, so a reader can pool connections

        def log_message(self, *args):
            pass

        def _send(self, status, body, ctype="application/json", extra=None):
            self.send_response(status)
            self.send_header("Content-Type", ctype)
            self.send_header("Content-Length", str(len(body)))
            for key, value in (extra or {}).items():
                self.send_header(key, value)
            self.end_headers()
            if self.command != "HEAD":
                self.wfile.write(body)

        def do_GET(self):
            try:
                length = int(self.headers.get("Content-Length") or 0)
                self.rfile.read(length)       # drain first: a refusal must not strand a body
                if not secrets.compare_digest(self.headers.get("X-Token", ""), token):
                    raise ChannelError("unauthorized", "this channel needs its token", 401)
                url = urlsplit(self.path)
                verb, _, rest = url.path.lstrip("/").partition("/")
                path = root / clean_rel(rest)
                if not path.is_file():
                    raise ChannelError("not_found", f"no file {rest!r}", 404)
                st = path.stat()
                tag = validator(st)
                if verb == "hash":            # the source's proof: one SHA-256 per chunk
                    shas = []
                    with open(path, "rb") as f:
                        while True:
                            block = f.read(CHUNK)
                            if not block:
                                break
                            shas.append(hashlib.sha256(block).hexdigest())
                    doc = {"size": st.st_size, "etag": tag, "chunks": shas}
                    return self._send(200, json.dumps(doc).encode())
                if verb != "fs":
                    raise ChannelError("not_found", f"no verb {verb!r}", 404)
                if self.headers.get("If-None-Match") == tag:
                    return self._send(304, b"", extra={"ETag": tag})
                with open(path, "rb") as f:
                    rng = self.headers.get("Range", "")
                    if rng.startswith("bytes="):
                        start, _, end = rng[6:].partition("-")
                        f.seek(int(start))
                        body = f.read(int(end) - int(start) + 1)
                        return self._send(206, body, "application/octet-stream", {"ETag": tag})
                    return self._send(200, f.read(), "application/octet-stream", {"ETag": tag})
            except ChannelError as exc:
                doc = {"error": exc.cause, "detail": exc.detail, "status": exc.status}
                self._send(exc.status, json.dumps(doc).encode())

    return Owner


class Reader:
    """The other end: a validator cache, named errors, resumable verified pulls."""

    def __init__(self, port, token, state):
        self.port, self.token, self.state = port, token, Path(state)
        self.cache = {}                   # rel -> (etag, bytes); Delta keeps blobs on disk

    def _get(self, path, headers=None):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=30)
        try:
            conn.request("GET", path, headers=dict(headers or {}, **{"X-Token": self.token}))
            resp = conn.getresponse()
            expected = resp.getheader("Content-Length")
            try:
                data = resp.read()
            except http.client.IncompleteRead as exc:   # the body ended before its length
                data = exc.partial
        except (OSError, http.client.HTTPException) as exc:
            raise ChannelError("unavailable", f"request to 127.0.0.1:{self.port} failed: {exc!r}", 503)
        finally:
            conn.close()
        if expected is not None and int(expected) != len(data):
            raise ChannelError("unavailable", f"closed mid-body at {len(data)} of {expected}", 502)
        if resp.status >= 400:
            doc = json.loads(data)
            raise ChannelError(doc["error"], doc["detail"], doc["status"])
        return resp.status, resp.getheader("ETag", ""), data

    def read(self, rel):
        """One file through the cache: 'miss' fetches, 'revalidated' costs a 304."""
        held = self.cache.get(rel)
        status, tag, data = self._get("/fs/" + quote(rel), {"If-None-Match": held[0]} if held else None)
        if status == 304:
            return held[1], "revalidated"
        self.cache[rel] = (tag, data)
        return data, "miss"

    def pull(self, rel, dest, stop_after=None):
        """Resumable: each chunk is checked, written, fsynced, THEN recorded."""
        proof = json.loads(self._get("/hash/" + quote(rel))[2])
        dest = Path(dest)
        part = self.state / (hashlib.sha256(f"{rel}{proof['etag']}".encode()).hexdigest()[:16] + ".part")
        record = part.with_suffix(".chunks")  # the key holds the version: a new one starts fresh
        done = set(record.read_text().split()) if record.exists() else set()
        if done:
            print(f"  resuming: {len(done)} of {len(proof['chunks'])} chunks already verified")
        with open(part, "r+b" if part.exists() else "w+b") as out:   # never "a": append ignores seek
            for i, sha in enumerate(proof["chunks"]):
                if str(i) in done:
                    continue
                if stop_after is not None and i >= stop_after:
                    raise ChannelError("unavailable", f"cut after chunk {i} (simulated)", 502)
                rng = f"bytes={i * CHUNK}-{min((i + 1) * CHUNK, proof['size']) - 1}"
                block = self._get("/fs/" + quote(rel), {"Range": rng})[2]
                if hashlib.sha256(block).hexdigest() != sha:
                    raise ChannelError("integrity", f"chunk {i} of {rel} failed its SHA-256", 502)
                out.seek(i * CHUNK)
                out.write(block)
                out.flush()
                os.fsync(out.fileno())
                with open(record, "a") as rec:   # one appended line per chunk
                    rec.write(f"{i}\n")
        whole, landed = hashlib.sha256(), []
        with open(part, "rb") as f:          # the file lands when its chunk list is the source's
            for block in iter(lambda: f.read(CHUNK), b""):
                whole.update(block)
                landed.append(hashlib.sha256(block).hexdigest())
        if landed != proof["chunks"] or os.path.getsize(part) != proof["size"]:
            raise ChannelError("integrity", f"{rel} does not match the source's chunk list", 502)
        os.replace(part, dest)                # atomic: the file lands whole or not at all
        record.unlink()
        return whole.hexdigest()


def main():
    with tempfile.TemporaryDirectory() as tmp:
        share, state = Path(tmp, "share"), Path(tmp, "state")
        share.mkdir()
        state.mkdir()
        (share / "big.bin").write_bytes(os.urandom(5 * CHUNK + 1234))
        (share / ".ssh").mkdir()
        (share / ".ssh" / "id_ed25519").write_text("never served")
        token = secrets.token_urlsafe(32)
        server = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(share, token))
        threading.Thread(target=server.serve_forever, daemon=True).start()
        reader = Reader(server.server_address[1], token, state)

        print("1. read twice:", reader.read("big.bin")[1], "then", reader.read("big.bin")[1])
        try:
            reader.pull("big.bin", Path(tmp, "big.copy"), stop_after=2)
        except ChannelError as exc:
            print("2. first pull stopped:", exc)
        digest = reader.pull("big.bin", Path(tmp, "big.copy"))
        same = digest == hashlib.sha256((share / "big.bin").read_bytes()).hexdigest()
        print("3. second pull landed, whole-file SHA-256 matches:", same)
        for rel in ("../etc/hosts", ".ssh/id_ed25519", "missing.txt"):
            try:
                reader.read(rel)
            except ChannelError as exc:
                print(f"4. {rel!r} ->", exc)
        reader.token = "wrong"
        try:
            reader.read("big.bin")
        except ChannelError as exc:
            print("5. wrong token ->", exc)

        class Cutter(make_handler(share, token)):
            """An owner that promises the whole file and closes halfway."""
            def _send(self, status, body, ctype="application/json", extra=None):
                if ctype != "application/octet-stream":
                    return super()._send(status, body, ctype, extra)
                self.send_response(status)
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body[: len(body) // 2])
                self.close_connection = True

        cutter = ThreadingHTTPServer(("127.0.0.1", 0), Cutter)
        threading.Thread(target=cutter.serve_forever, daemon=True).start()
        try:
            Reader(cutter.server_address[1], token, state).read("big.bin")
        except ChannelError as exc:
            print("6. owner cut the body ->", exc)
        cutter.shutdown()
        server.shutdown()


if __name__ == "__main__":
    main()

What the run prints·text

1. read twice: miss then revalidated
2. first pull stopped: unavailable: cut after chunk 2 (simulated)
  resuming: 2 of 6 chunks already verified
3. second pull landed, whole-file SHA-256 matches: True
4. '../etc/hosts' -> forbidden: '..' and NUL are not path components
4. '.ssh/id_ed25519' -> forbidden: .ssh/id_ed25519 is inside a refused place
4. 'missing.txt' -> not_found: no file 'missing.txt'
5. wrong token -> unauthorized: this channel needs its token
6. owner cut the body -> unavailable: closed mid-body at 2622057 of 5244114

External links

Exercise

Run minichannel.py, then extend it in three steps, keeping the demo green after each: (1) persist the reader's cache to disk as files named by SHA-256, with an index from path to validator and blob; (2) pull with two worker threads sharing one record, and prove a resume still works when chunks land out of order; (3) change the source file between the cut and the second pull and show that the second pull starts a fresh partial and lands the new version, leaving the old partial for a sweep you write.

Hint

For (2), the record must be shared under one lock per partial (Delta learned this from a 40 MiB push whose commit was refused with "2 of 3 chunks still missing": each request had kept its own record and saved it over the others'), and the partial must be opened r+b. For (3), list the state folder before and after: the partial's name contains the validator, so a new version gets a new name.