diff --git a/minigit/remote.py b/minigit/remote.py index 566ef2f..883ad6f 100644 --- a/minigit/remote.py +++ b/minigit/remote.py @@ -65,6 +65,33 @@ def recv_exact(sock, buf: bytearray, size: int) -> bytes: return data +def _valid_hash(value: str) -> bool: + """Return whether `value` is a full 40-character lowercase hex object hash.""" + + return len(value) == 40 and all(c in "0123456789abcdef" for c in value) + + +def _valid_branch(name: str) -> bool: + """Return whether `name` is safe to use as a ref path under `refs/heads/`. + + Rejects empty names, `.`/`..`/empty path segments (traversal), and + characters that would break a ref file or a protocol line. + """ + + return ( + bool(name) + and all(part not in {"", ".", ".."} for part in name.split("/")) + and not any(char in name for char in "\0\\\r\n ") + ) + + +def _rejection_reason(reply: str) -> str: + """Strip the `ERR ` prefix off a server reply so it reads well in a CLI error.""" + + command, _, reason = reply.partition(" ") + return reason if command == "ERR" and reason else reply + + def _has_object(store, obj_hash: str) -> bool: """Return whether `obj_hash` already exists in `store`. @@ -131,12 +158,14 @@ def push(self, remote_address: str, branch: str, token: str) -> None: """Send local commits on `branch` to the remote, rejecting if it has diverged. Resolves the local tip and authenticates, then reads the remote's - current tip through `REF`. Equal tips mean nothing to do; a remote - tip that is not an ancestor of the local tip means someone else - pushed first, and the push is rejected. Otherwise every object - reachable from `branch` that the remote doesn't already have is + current tip through `REF`. Equal tips mean nothing to do. A remote + tip that is not an ancestor of the local tip (unknown locally, ahead + of local, or diverged) means someone else pushed first, and the push + is rejected before anything is uploaded. Otherwise every object + reachable from the local tip that the remote doesn't already have is uploaded (`HAVE`/`PUT`), and the ref only moves after `DONE` reports - the transfer validated on the far side. + the transfer validated on the far side - the server rechecks the + expected old tip there too, so a push that raced ours still loses. """ host, port = self._parse_address(remote_address) @@ -165,27 +194,25 @@ def push(self, remote_address: str, branch: str, token: str) -> None: if len(fields) != 3 or fields[:2] != ["REF", branch]: raise NetworkProtocolError(f"invalid REF response: {reply!r}") remote_hash = fields[2] - if remote_hash != "-" and ( - len(remote_hash) != 40 or any(c not in "0123456789abcdef" for c in remote_hash) - ): + if remote_hash != "-" and not _valid_hash(remote_hash): raise NetworkProtocolError(f"invalid remote hash: {remote_hash!r}") if remote_hash == local_hash: + send_line(sock, "DONE") print(f"{branch} is up to date") return - if remote_hash != "-" and not _is_ancestor(self.commits, remote_hash, local_hash): - raise NetworkProtocolError( - f"remote {branch} has diverged from local: " - f"{remote_hash} is not an ancestor of {local_hash}" - ) + if remote_hash != "-": + self._check_fast_forward(branch, remote_hash, local_hash) send_line(sock, f"PUSH {branch} {remote_hash} {local_hash}") reply = receive_line(sock, buf) if reply != "OK": - raise NetworkProtocolError(f"push rejected: {reply}") + raise NetworkProtocolError(f"push rejected: {_rejection_reason(reply)}") - for obj_hash in self.collect_reachable(branch): + # Pinned to `local_hash`, not re-read from the ref: a local commit + # landing mid-push must not change the snapshot we proposed. + for obj_hash in self._collect_from(local_hash): send_line(sock, f"HAVE {obj_hash}") reply = receive_line(sock, buf) if reply == "NO": @@ -196,7 +223,7 @@ def push(self, remote_address: str, branch: str, token: str) -> None: send_line(sock, "DONE") reply = receive_line(sock, buf) if reply != "OK": - raise NetworkProtocolError(f"push failed: {reply}") + raise NetworkProtocolError(f"push rejected: {_rejection_reason(reply)}") print(f"{branch} now at {local_hash}") except NetworkProtocolError: @@ -206,6 +233,29 @@ def push(self, remote_address: str, branch: str, token: str) -> None: finally: sock.close() + def _check_fast_forward(self, branch: str, remote_hash: str, local_hash: str) -> None: + """Raise unless moving the remote from `remote_hash` to `local_hash` is a fast-forward. + + Ancestry comes from M3's history walk, which follows every parent of + a merge commit, so a remote tip reachable only through a merge's + second parent still counts as an ancestor. + """ + + if _is_ancestor(self.commits, remote_hash, local_hash): + return + if not _has_object(self.store, remote_hash): + raise NetworkProtocolError( + f"push rejected: remote {branch} is at {remote_hash[:7]}, " + "which you don't have locally - pull first" + ) + if _is_ancestor(self.commits, local_hash, remote_hash): + raise NetworkProtocolError( + f"push rejected: remote {branch} is ahead of local - pull first" + ) + raise NetworkProtocolError( + f"push rejected: local and remote {branch} have diverged - pull and merge first" + ) + def _put_object(self, sock, buf: bytearray, obj_hash: str) -> None: """Send `PUT ` followed by the object's `OBJ` header and bytes.""" @@ -304,15 +354,15 @@ def _receive_object(self, sock, buf: bytearray, expected_hash: str) -> None: self.store.write_object(content, obj_type) def collect_reachable(self, branch: str) -> set[str]: - """Return every commit, tree, and blob hash reachable from `branch`'s tip. - - Local-only this week: this is what push will later diff against the - remote's advertised hash to find what's actually missing there. - """ + """Return every commit, tree, and blob hash reachable from `branch`'s tip.""" tip = self.commits.read_ref(branch) if tip is None: return set() + return self._collect_from(tip) + + def _collect_from(self, tip: str) -> set[str]: + """Return every commit, tree, and blob hash reachable from commit `tip`.""" reachable: set[str] = set() for commit_hash in self.commits.walk_history(tip): @@ -337,12 +387,34 @@ def _collect_tree(self, tree_hash: str, reachable: set[str]) -> None: class RemoteServer: - """Accepts a RemoteClient's AUTH + REF handshake over TCP, one client at a time.""" + """Serves one repo's refs and objects to `RemoteClient`s over TCP, one client at a time. + + Sessions are handled strictly one after another, which is what makes a + push's final "recheck ref -> validate history -> write ref" sequence + atomic with respect to other pushes: no second session can run between + the check and the write. If `serve_forever` is ever made concurrent, that + sequence in `_handle_push` must be wrapped in a lock. (Writes made to the + served repo outside this server, e.g. a local commit, aren't covered.) + + Each accepted connection gets `conn_timeout` seconds per read/write, so a + stalled client can't hold the server - and every client queued behind + it - forever. + """ - def __init__(self, repo_path=".", token="", host="127.0.0.1", port=0, store=None, commits=None): + def __init__( + self, + repo_path=".", + token="", + host="127.0.0.1", + port=0, + store=None, + commits=None, + conn_timeout=10.0, + ): self.repo_path = repo_path self.token = token self.host = host + self.conn_timeout = conn_timeout self.store = store if store is not None else ObjectStore(repo_path) self.commits = ( commits if commits is not None else CommitManager(repo_path, store=self.store) @@ -377,6 +449,7 @@ def serve_forever(self) -> None: return # listening socket was closed - shut down try: + conn.settimeout(self.conn_timeout) self._handle_client(conn) except (ObjectNotFoundError, ObjectCorruptError) as exc: try: @@ -420,11 +493,7 @@ def _handle_client(self, conn) -> None: def _send_ref(self, conn, branch: str) -> None: """Reply with the commit hash `branch` currently points at, or `-` if it has none.""" - if ( - not branch - or any(part in {"", ".", ".."} for part in branch.split("/")) - or any(char in branch for char in "\0\\\r\n") - ): + if not _valid_branch(branch): send_line(conn, "ERR invalid branch") return commit_hash = self.commits.read_ref(branch) or "-" @@ -445,20 +514,19 @@ def _handle_push(self, conn, buf: bytearray, args: str) -> None: send_line(conn, "ERR malformed PUSH") return - if ( - not branch - or any(part in {"", ".", ".."} for part in branch.split("/")) - or any(char in branch for char in "\0\\\r\n") - ): + if not _valid_branch(branch): send_line(conn, "ERR invalid branch") return - if len(new_hash) != 40 or any(c not in "0123456789abcdef" for c in new_hash): + if old_hash != "-" and not _valid_hash(old_hash): + send_line(conn, "ERR invalid old hash") + return + if not _valid_hash(new_hash): send_line(conn, "ERR invalid new hash") return expected_old = None if old_hash == "-" else old_hash if self.commits.read_ref(branch) != expected_old: - send_line(conn, f"ERR {branch} changed: expected {old_hash}") + send_line(conn, f"ERR remote {branch} has moved since you last synced - pull first") return send_line(conn, "OK") @@ -468,6 +536,9 @@ def _handle_push(self, conn, buf: bytearray, args: str) -> None: command, _, value = line.partition(" ") if command == "HAVE": + if not _valid_hash(value): + send_line(conn, "ERR invalid object hash") + return send_line(conn, "YES" if _has_object(self.store, value) else "NO") elif command == "PUT": if not self._receive_pushed_object(conn, buf, value): @@ -478,10 +549,11 @@ def _handle_push(self, conn, buf: bytearray, args: str) -> None: send_line(conn, "ERR expected HAVE, PUT, or DONE") return - # Recheck the ref hasn't moved since we accepted the PUSH - another - # client's push could have landed while we were receiving objects. + # Recheck the ref hasn't moved since we accepted the PUSH. From here + # to write_ref below nothing else can move it: sessions are serial + # (see the class docstring), so check + write act as one step. if self.commits.read_ref(branch) != expected_old: - send_line(conn, "ERR ref changed during push") + send_line(conn, f"ERR remote {branch} changed during push - pull first") return try: @@ -505,7 +577,7 @@ def _receive_pushed_object(self, conn, buf: bytearray, expected_hash: str) -> bo session on failure. """ - if len(expected_hash) != 40 or any(c not in "0123456789abcdef" for c in expected_hash): + if not _valid_hash(expected_hash): send_line(conn, "ERR invalid object hash") return False @@ -559,7 +631,7 @@ def _validate_tree(self, tree_hash: str, visited: set[str]) -> None: def _send_object(self, conn, obj_hash: str) -> None: """Reply with the requested object's bytes, or `ERR` if it isn't in the store.""" - if len(obj_hash) != 40 or any(c not in "0123456789abcdef" for c in obj_hash): + if not _valid_hash(obj_hash): send_line(conn, "ERR invalid object hash") return try: @@ -592,6 +664,12 @@ def _send_object(self, conn, obj_hash: str) -> None: # (propose the move) -> repeated HAVE/PUT (upload only what the remote is # missing) -> DONE (remote validates the new history and moves the ref) -> # connection closes. +# +# Conflict detection happens twice. The client refuses before PUSH if the +# remote tip from REF isn't an ancestor of its local tip. The server +# refuses PUSH if isn't its current ref, and again at DONE if the ref +# moved, history is incomplete, or doesn't descend from . Any +# refusal is `ERR ` and leaves the ref where it was. def register_subcommands(subparsers) -> None: diff --git a/tests/test_remote.py b/tests/test_remote.py index 97be842..a4d4b01 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -7,6 +7,7 @@ import pytest +from minigit.cli import main from minigit.commits import CommitManager from minigit.errors import NetworkProtocolError from minigit.objects import ObjectStore @@ -326,11 +327,12 @@ def test_push_disconnect_before_done_leaves_ref_unchanged(push_pair): def test_push_rejects_unrelated_remote_tip(push_pair): pair = push_pair unrelated = _commit(pair.remote_store, pair.remote_commits, {"other.txt": b"x"}, []) - _commit(pair.local_store, pair.local_commits, {"a.txt": b"1"}, []) + local_tip = _commit(pair.local_store, pair.local_commits, {"a.txt": b"1"}, []) - with pytest.raises(NetworkProtocolError): + with pytest.raises(NetworkProtocolError, match="don't have locally - pull first"): pair.client.push(pair.address, "main", "tok") assert pair.remote_commits.read_ref("main") == unrelated + assert pair.local_commits.read_ref("main") == local_tip def test_push_rejects_ref_changed_during_transfer(push_pair): @@ -734,3 +736,286 @@ def close(self): monkeypatch.setattr(socket, "create_connection", lambda *args, **kwargs: sock) with pytest.raises(NetworkProtocolError, match="invalid"): RemoteClient(tmp_path).push("localhost:9418", "main", "tok") + + +# --- push conflict detection (#34) --- + + +def _second_client(tmp_path, pair: PushPair) -> RemoteClient: + """Another independent local repo, pointed at the same remote as `pair.client`.""" + + root = tmp_path / "local2" + root.mkdir() + store = ObjectStore(str(root)) + return RemoteClient( + repo_path=str(root), store=store, commits=CommitManager(str(root), store=store) + ) + + +def _count_puts(monkeypatch) -> list: + """Record every object `RemoteClient` uploads, while still uploading it.""" + + calls = [] + original_put = RemoteClient._put_object + + def counting_put(self, sock, buf, obj_hash): + calls.append(obj_hash) + return original_put(self, sock, buf, obj_hash) + + monkeypatch.setattr(RemoteClient, "_put_object", counting_put) + return calls + + +def _open_push(pair: PushPair, old: str, new: str) -> tuple[socket.socket, bytearray]: + """Authenticate a raw connection and send `PUSH main `; returns the reply buffer.""" + + sock = socket.create_connection(("127.0.0.1", pair.server.port), timeout=5) + buf = bytearray() + send_line(sock, "AUTH tok") + assert receive_line(sock, buf) == "OK" + send_line(sock, f"PUSH main {old} {new}") + return sock, buf + + +def _upload(sock, buf: bytearray, store: ObjectStore, hashes) -> None: + """Raw-protocol HAVE/PUT for each hash, uploading whatever the remote is missing.""" + + for obj_hash in hashes: + send_line(sock, f"HAVE {obj_hash}") + if receive_line(sock, buf) == "NO": + obj_type, content = store.read_object(obj_hash) + send_line(sock, f"PUT {obj_hash}") + send_line(sock, f"OBJ {obj_type} {len(content)}") + sock.sendall(content) + assert receive_line(sock, buf) == "OK" + + +def test_push_fast_forward_then_repeat_transfers_nothing(push_pair, monkeypatch, capsys): + pair = push_pair + c1 = _commit(pair.local_store, pair.local_commits, {"a.txt": b"1"}, []) + pair.client.push(pair.address, "main", "tok") + c2 = _commit(pair.local_store, pair.local_commits, {"a.txt": b"2"}, [c1]) + pair.client.push(pair.address, "main", "tok") + assert pair.remote_commits.read_ref("main") == c2 + + puts = _count_puts(monkeypatch) + capsys.readouterr() + pair.client.push(pair.address, "main", "tok") + assert puts == [] + assert "up to date" in capsys.readouterr().out + assert pair.remote_commits.read_ref("main") == c2 + + +def test_push_accepts_ancestor_only_via_second_merge_parent(push_pair): + pair = push_pair + remote_tip = _commit(pair.local_store, pair.local_commits, {"a.txt": b"a"}, []) + pair.client.push(pair.address, "main", "tok") + + # `other` is an unrelated root, so the remote tip is reachable only through + # the merge commit's *second* parent - a first-parent-only walk would miss it. + other = _commit(pair.local_store, pair.local_commits, {"b.txt": b"b"}, []) + merge = _commit( + pair.local_store, pair.local_commits, {"a.txt": b"a", "b.txt": b"b"}, [other, remote_tip] + ) + + pair.client.push(pair.address, "main", "tok") + assert pair.remote_commits.read_ref("main") == merge + assert set(pair.remote_commits.walk_history(merge)) == {merge, other, remote_tip} + + +def test_push_rejects_remote_ahead(push_pair): + pair = push_pair + c1 = _commit(pair.local_store, pair.local_commits, {"a.txt": b"1"}, []) + c2 = _commit(pair.local_store, pair.local_commits, {"a.txt": b"2"}, [c1]) + pair.client.push(pair.address, "main", "tok") + pair.local_commits.write_ref("main", c1) # local falls behind a remote it has seen + + with pytest.raises(NetworkProtocolError, match="ahead of local - pull first"): + pair.client.push(pair.address, "main", "tok") + assert pair.remote_commits.read_ref("main") == c2 + assert pair.local_commits.read_ref("main") == c1 + + +def test_push_rejects_diverged(push_pair): + pair = push_pair + base = _commit(pair.local_store, pair.local_commits, {"a.txt": b"base"}, []) + theirs = _commit(pair.local_store, pair.local_commits, {"a.txt": b"theirs"}, [base]) + pair.client.push(pair.address, "main", "tok") + ours = _commit(pair.local_store, pair.local_commits, {"a.txt": b"ours"}, [base]) + + with pytest.raises(NetworkProtocolError, match="diverged - pull and merge first"): + pair.client.push(pair.address, "main", "tok") + assert pair.remote_commits.read_ref("main") == theirs + assert pair.local_commits.read_ref("main") == ours + + +def test_two_stale_clients_second_rejected(tmp_path, push_pair): + """Week 5 checkpoint: two repos share a remote tip, commit independently, push both.""" + + pair = push_pair + alice = pair.client + bob = _second_client(tmp_path, pair) + + base = _commit(pair.local_store, pair.local_commits, {"a.txt": b"base"}, []) + alice.push(pair.address, "main", "tok") + bob.fetch_objects(pair.address, sorted(alice.collect_reachable("main")), "tok") + bob.commits.write_ref("main", base) + + alice_tip = _commit(pair.local_store, pair.local_commits, {"a.txt": b"alice"}, [base]) + bob_tip = _commit(bob.store, bob.commits, {"a.txt": b"bob"}, [base]) + + alice.push(pair.address, "main", "tok") + with pytest.raises(NetworkProtocolError, match="pull first"): + bob.push(pair.address, "main", "tok") + + assert pair.remote_commits.read_ref("main") == alice_tip + assert bob.commits.read_ref("main") == bob_tip + + +def test_stale_push_rejected_by_server(push_pair): + """A client that skips its own ancestry check still can't overwrite a moved ref.""" + + pair = push_pair + base = _commit(pair.local_store, pair.local_commits, {"a.txt": b"base"}, []) + pair.client.push(pair.address, "main", "tok") + moved = _commit(pair.remote_store, pair.remote_commits, {"a.txt": b"moved"}, [base]) + stale = _commit(pair.local_store, pair.local_commits, {"a.txt": b"stale"}, [base]) + + sock, buf = _open_push(pair, base, stale) + try: + reply = receive_line(sock, buf) + finally: + sock.close() + + assert reply == "ERR remote main has moved since you last synced - pull first" + assert pair.remote_commits.read_ref("main") == moved + + +def test_push_rejects_existing_ref_moved_during_transfer(push_pair): + pair = push_pair + base = _commit(pair.local_store, pair.local_commits, {"a.txt": b"base"}, []) + pair.client.push(pair.address, "main", "tok") + new = _commit(pair.local_store, pair.local_commits, {"a.txt": b"new"}, [base]) + + sock, buf = _open_push(pair, base, new) + try: + assert receive_line(sock, buf) == "OK" + pair.remote_commits.write_ref("main", "b" * 40) # another push lands mid-transfer + _upload(sock, buf, pair.local_store, sorted(pair.client.collect_reachable("main"))) + send_line(sock, "DONE") + assert receive_line(sock, buf) == "ERR remote main changed during push - pull first" + finally: + sock.close() + + assert pair.remote_commits.read_ref("main") == "b" * 40 + + +def test_direct_push_not_fast_forward(push_pair): + pair = push_pair + base = _commit(pair.local_store, pair.local_commits, {"a.txt": b"base"}, []) + pair.client.push(pair.address, "main", "tok") + unrelated = _commit(pair.local_store, pair.local_commits, {"b.txt": b"other"}, []) + + sock, buf = _open_push(pair, base, unrelated) + try: + assert receive_line(sock, buf) == "OK" + _upload(sock, buf, pair.local_store, sorted(pair.client.collect_reachable("main"))) + send_line(sock, "DONE") + assert receive_line(sock, buf) == "ERR not a fast-forward" + finally: + sock.close() + + assert pair.remote_commits.read_ref("main") == base + + +def test_direct_push_missing_objects(push_pair): + pair = push_pair + tip = _commit(pair.local_store, pair.local_commits, {"a.txt": b"1"}, []) + + sock, buf = _open_push(pair, "-", tip) + try: + assert receive_line(sock, buf) == "OK" + send_line(sock, "DONE") # claims done without uploading anything + assert receive_line(sock, buf).startswith("ERR incomplete history") + finally: + sock.close() + + assert pair.remote_commits.read_ref("main") is None + + +@pytest.mark.parametrize("old", ["zzz", "A" * 40, "a" * 39]) +def test_direct_push_invalid_old_hash(push_pair, old): + pair = push_pair + tip = _commit(pair.local_store, pair.local_commits, {"a.txt": b"1"}, []) + + sock, buf = _open_push(pair, old, tip) + try: + assert receive_line(sock, buf) == "ERR invalid old hash" + finally: + sock.close() + + assert pair.remote_commits.read_ref("main") is None + + +def test_retry_after_disconnect_reuses_objects(push_pair, monkeypatch): + pair = push_pair + tip = _commit( + pair.local_store, + pair.local_commits, + {"a.txt": b"1", "b.txt": b"2", "dir/c.txt": b"3"}, + [], + ) + objects = sorted(pair.client.collect_reachable("main")) + + sock, buf = _open_push(pair, "-", tip) + try: + assert receive_line(sock, buf) == "OK" + _upload(sock, buf, pair.local_store, objects[:2]) + finally: + sock.close() # drop the connection before DONE + assert pair.remote_commits.read_ref("main") is None + + puts = _count_puts(monkeypatch) + pair.client.push(pair.address, "main", "tok") + assert pair.remote_commits.read_ref("main") == tip + assert sorted(puts) == objects[2:] + + +def test_stalled_client_does_not_block_server(tmp_path): + server = RemoteServer(repo_path=str(tmp_path), token="tok", port=0, conn_timeout=0.5) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + stalled = socket.create_connection(("127.0.0.1", server.port), timeout=5) + try: + buf = bytearray() + send_line(stalled, "AUTH tok") + assert receive_line(stalled, buf) == "OK" + send_line(stalled, f"PUSH main - {'a' * 40}") + assert receive_line(stalled, buf) == "OK" + # ...and then goes silent mid-session, holding the serial server. + + with socket.create_connection(("127.0.0.1", server.port), timeout=3) as sock: + buf = bytearray() + send_line(sock, "AUTH tok") + assert receive_line(sock, buf) == "OK" + send_line(sock, "REF main") + assert receive_line(sock, buf) == "REF main -" + send_line(sock, "DONE") + finally: + stalled.close() + server.close() + thread.join(timeout=2) + assert not thread.is_alive() + + +def test_cli_push_rejection_shows_reason(push_pair, monkeypatch, capsys): + pair = push_pair + _commit(pair.remote_store, pair.remote_commits, {"other.txt": b"x"}, []) + _commit(pair.local_store, pair.local_commits, {"a.txt": b"1"}, []) + monkeypatch.chdir(pair.client.repo_path) + + assert main(["push", pair.address, "main", "--token", "tok"]) == 1 + err = capsys.readouterr().err + assert "push rejected" in err + assert "pull first" in err + assert "tok" not in err