"""Local HTTP conditional-write lab; Python 3.10+, standard library only. Runs its own loopback server, exercises it, then closes it. Not a production server or a complete implementation of all HTTP precondition combinations. """ from concurrent.futures import ThreadPoolExecutor from http.client import HTTPConnection from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import platform import re import threading class Note: def __init__(self): self.lock = threading.Lock() self.body = b"original" self.revision = 1 def etag(self): return '"v%d"' % self.revision class Handler(BaseHTTPRequestHandler): # Deliberate teaching subset: ONE quoted ASCII tag, optionally W/. # Lists, wildcard, combined fields and other preconditions are rejected. protocol_version = "HTTP/1.1" server_version = "ConditionalWriteLab/1" def setup(self): super().setup() self.connection.settimeout(3) def log_message(self, *args): pass def reply(self, status, body, etag=None): self.send_response(status) self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("Connection", "close") if etag is not None: self.send_header("ETag", etag) self.end_headers() self.wfile.write(body) self.close_connection = True def allowed_path(self): if self.path not in self.server.notes: self.reply(404, b"unknown resource") return False return True def do_GET(self): if not self.allowed_path(): return if any(name.lower().startswith("if-") for name in self.headers): self.reply(400, b"conditional GET is outside this lab") return note = self.server.notes[self.path] with note.lock: body, etag = note.body, note.etag() self.reply(200, body, etag) def do_PUT(self): if not self.allowed_path(): return if self.headers.get_content_type() != "text/plain": self.reply(415, b"send text/plain UTF-8") return if self.headers.get("Transfer-Encoding") is not None: self.reply(400, b"transfer encoding is outside this lab") return lengths = self.headers.get_all("Content-Length", []) if len(lengths) != 1 or not re.fullmatch(r"[0-9]+", lengths[0]): self.reply(400, b"one numeric Content-Length is required") return size = int(lengths[0]) if size > 2048: self.reply(413, b"maximum 2048 bytes") return if any(name.lower().startswith("if-") and name.lower() != "if-match" for name in self.headers): self.reply(400, b"other preconditions are outside this lab") return tags = self.headers.get_all("If-Match", []) if len(tags) > 1 or (tags and not re.fullmatch(r'(?:W/)?"[!#-~]*"', tags[0])): self.reply(400, b"lab accepts one quoted ASCII entity tag only") return if self.path == "/note" and not tags: self.reply(428, b"read /note and resend with its ETag in If-Match") return try: body = self.rfile.read(size) body.decode("utf-8") except (UnicodeDecodeError, TimeoutError): self.reply(400, b"complete UTF-8 body required") return if len(body) != size: self.reply(400, b"incomplete body") return note = self.server.notes[self.path] # The comparison and write share ONE lock. Moving them into two # independent critical sections reintroduces a lost-update window. with note.lock: if tags and (tags[0].startswith("W/") or tags[0] != note.etag()): result = (412, b"stale or weak validator; reread before merging", None) else: note.body = body # Store the exact received bytes, no transformation. note.revision += 1 result = (200, body, note.etag()) self.reply(*result) def run(): server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) server.notes = {"/note": Note(), "/unguarded": Note()} thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() rows = [] def request(label, method, path="/note", body=None, tag=None): headers = {"Content-Type": "text/plain; charset=utf-8"} if tag is not None: headers["If-Match"] = tag conn = HTTPConnection("127.0.0.1", server.server_port, timeout=5) try: conn.request(method, path, body=body, headers=headers) response = conn.getresponse() data = response.read().decode("utf-8") result = {"label": label, "status": response.status, "etag": response.getheader("ETag"), "body": data} rows.append(result) return result finally: conn.close() try: a = request("A reads", "GET") b = request("B reads", "GET") assert a["etag"] == b["etag"] == '"v1"' accepted = request("A writes", "PUT", body=b"A edited", tag=a["etag"]) assert accepted["status"] == 200 and accepted["etag"] == '"v2"' stale = request("B stale write", "PUT", body=b"B overwrites", tag=b["etag"]) assert stale["status"] == 412 current = request("read after conflict", "GET") assert current["body"] == "A edited" and current["etag"] == '"v2"' assert request("missing condition", "PUT", body=b"unconditional")["status"] == 428 assert request("weak condition", "PUT", body=b"weak", tag='W/"v2"')["status"] == 412 assert request("unsupported wildcard", "PUT", body=b"wildcard", tag="*")["status"] == 400 assert request("invalid UTF-8", "PUT", body=b"\xff", tag='"v2"')["status"] == 400 unchanged = request("read after rejected writes", "GET") assert unchanged["etag"] == '"v2"' and unchanged["body"] == "A edited" merged = request("B reviewed merge", "PUT", body=b"A edited + B addition", tag=unchanged["etag"]) assert merged["status"] == 200 and merged["etag"] == '"v3"' # Two concurrent requests with the same ETag: exactly one can commit. gate = threading.Barrier(2) def competing(value): gate.wait(timeout=5) return request("concurrent " + value, "PUT", body=value.encode(), tag='"v3"') with ThreadPoolExecutor(max_workers=2) as pool: outcomes = list(pool.map(competing, ("C edit", "D edit"))) assert sorted(item["status"] for item in outcomes) == [200, 412] winner = next(item for item in outcomes if item["status"] == 200) after = request("read after race", "GET") assert after["body"] == winner["body"] and after["etag"] == '"v4"' # Counterexample: a separate resource explicitly permits blind overwrite. request("unguarded A", "PUT", path="/unguarded", body=b"A edited") request("unguarded B", "PUT", path="/unguarded", body=b"B overwrites") assert request("unguarded final", "GET", path="/unguarded")["body"] == "B overwrites" assert request("unknown route", "GET", path="/missing")["status"] == 404 return {"python": platform.python_version(), "passed": True, "scope": "real loopback HTTP requests; in-memory single-process resource", "requests": rows} finally: server.shutdown() server.server_close() thread.join(timeout=5) assert not thread.is_alive(), "server thread failed to stop" if __name__ == "__main__": if not __debug__: raise SystemExit("Run without -O: this lab uses assertions to execute and verify requests.") print(json.dumps(run(), ensure_ascii=False, indent=2))