Loopback Is Not Identity

# delta/server.py and delta/identity.py (trimmed excerpt)

def admit(self, sock, client_address):             # Listener: classify a new connection
    peer_ip = client_address[0]
    if self.role == "engine":
        if not ipaddress.ip_address(peer_ip).is_loopback:
            log.warning("engine refused a non-loopback caller %s", peer_ip)
            return None
        return Conn("engine", peer_ip, local_ip, sock, self.delta.label)
    if ipaddress.ip_address(peer_ip).is_loopback:
        # A process on this Mac: this Mac's label, once it shows the local token
        # (the engine's rule; without it any account here read the home share).
        return Conn("loopback", peer_ip, local_ip, sock)
    ...


def _local_token_ok(self) -> bool:
    token = self.headers.get("X-Delta-Local-Token", "")
    return bool(token) and bool(self.delta.local_token) and \
        hmac.compare_digest(token.encode(), self.delta.local_token.encode())


def ensure_local_token() -> str:
    path = local_token_path()                      # <state root>/local.token
    try:
        token = path.read_text().strip()
    except FileNotFoundError:
        token = ""
    if token:
        return token
    # Absent, or empty (a stop between creating the file and writing it): an empty
    # token would refuse every app on this Mac for good, so it is written again,
    # whole, under a name of its own first.
    token = secrets.token_urlsafe(32)
    tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as handle:
        handle.write(token + "\n")
    os.replace(tmp, path)
    return token

#!/usr/bin/env python3
"""A loopback service that asks callers for a 0600 token file, the way Delta's engine does."""
import hmac
import http.client
import http.server
import os
import secrets
import tempfile
import threading
from pathlib import Path


def ensure_token(path: Path) -> str:
    try:
        token = path.read_text().strip()
    except FileNotFoundError:
        token = ""
    if token:
        return token
    # Missing or empty: write it whole under a temporary name, then rename into place.
    token = secrets.token_urlsafe(32)
    tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as handle:
        handle.write(token + "\n")
    os.replace(tmp, path)
    return token


class Handler(http.server.BaseHTTPRequestHandler):
    token = ""

    def do_GET(self):
        if self.path == "/health":                     # the one route anyone on loopback may ask
            return self.answer(200, b"ok\n")
        shown = self.headers.get("X-Local-Token", "")
        if not shown or not hmac.compare_digest(shown.encode(), self.token.encode()):
            return self.answer(401, b"show X-Local-Token (the 0600 file only this account can read)\n")
        self.answer(200, f"hello, {self.headers.get('X-App', '?')}\n".encode())

    def answer(self, status: int, body: bytes):
        self.send_response(status)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


with tempfile.TemporaryDirectory() as tmp:
    token_file = Path(tmp) / "local.token"
    token_file.write_text("")                          # the crash case: an empty file left behind
    Handler.token = ensure_token(token_file)
    print("mode", oct(token_file.stat().st_mode & 0o777), "length", len(Handler.token))

    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()

    def get(path, headers=None):
        conn = http.client.HTTPConnection("127.0.0.1", server.server_address[1], timeout=5)
        conn.request("GET", path, headers=headers or {})
        resp = conn.getresponse()
        out = resp.status, resp.read().decode().strip()
        conn.close()
        return out

    print(get("/health"))
    print(get("/data"))
    print(get("/data", {"X-Local-Token": token_file.read_text().strip(), "X-App": "my-app"}))
    server.shutdown()