diff --git a/README.md b/README.md
index bc5cc02d..2ed2dd25 100644
--- a/README.md
+++ b/README.md
@@ -16,7 +16,7 @@ The platform exposes four subsystems:
 
 - **Actions** — outbound SCM operations initiated by Sentry code. The `SourceCodeManager` class provides 70+ methods covering comments, reactions, pull requests, branches, git objects, reviews, and check runs. With more actions planned to be added as we port more use cases.
 - **Actions RPC** - outbound SCM commands exposed over the network. The `SourceCodeManager` is fully available over the network. This enables usage of the SCM for services outside the monolith.
-- **Git relay** - `git` fetches and pushes for services that hold no provider credentials. Seer signs a grant naming the organization, operation and repositories with `scm.git.grant.sign_grant` and an Ed25519 private key only Seer holds. `scm.git.server.GitServer` verifies the grant against the matching public keys, so whoever verifies grants cannot mint one, and checks each git smart-HTTP request against it before relaying the request to the provider with a token, streaming both bodies. A missing or unusable key on either side raises `scm.errors.GitGrantKeyError`, the server's misconfiguration rather than a refused grant. `require_private_key` and `require_public_keys` check the keys at startup, and `generate_private_key` makes a key for tests and local tools. `key_id` gives a public key's `kid`, the id each grant names its signing key by, so a log can name the key that signed a grant without quoting it.
+- **Git relay** - `git` fetches and pushes for services that hold no provider credentials. Seer signs a grant with `scm.git.grant.sign_grant` and an Ed25519 private key only Seer holds. A `Grant` names the organization and a list of rules, each a repository, an operation (`FetchRule` or `PushRule`) and the branches it may touch, or `any_object` for a fetch that may read any object. A push rule names its branches or full-match `branch_patterns`, and `create_only` limits it to new branches. A request passes if any rule allows it. `scm.git.server.GitServer` verifies the grant against the matching public keys, so whoever verifies grants cannot mint one, and checks each git smart-HTTP request against it before relaying the request to the provider with a token, streaming both bodies. For a push under a push rule it first reads the ref updates, and answers a push that updates any other ref, deletes a branch or carries push options with a status report git shows as rejected refs. A missing or unusable key on either side raises `scm.errors.GitGrantKeyError`, the server's misconfiguration rather than a refused grant. `require_private_key` and `require_public_keys` check the keys at startup, and `generate_private_key` makes a key for tests and local tools. `key_id` gives a public key's `kid`, the id each grant names its signing key by, so a log can name the key that signed a grant without quoting it.
 - **Event Stream** — inbound webhook processing. SCM service-providers push events which are deserialized into typed, provider-neutral dataclasses (`CheckRunEvent`, `CommentEvent`, `PullRequestEvent`) which are then dispatched to registered listener functions.
 
 ### Why
diff --git a/bin/git-server b/bin/git-server
index 6096ef24..54223dba 100755
--- a/bin/git-server
+++ b/bin/git-server
@@ -50,14 +50,15 @@ Configuration:
         GITLAB_BASE_URL                      Instance URL (default: https://gitlab.com)
 
 Serving:
-    The server prints a fetch grant and a push grant for the repository, valid
-    for an hour, with commands that pass them to git:
+    The server prints a grant to fetch any object of the repository and one to
+    push its branch my-branch, valid for an hour, with commands that pass them
+    to git:
 
         git -c http.extraHeader="X-Scm-Git-Grant: $FETCH_GRANT" \\
             clone http://127.0.0.1:8080/OWNER/REPO.git
 
     The URL path is the repository's path, which the server uses as its external
-    id, so a grant names the repository as (provider, path).
+    id, so a grant's rule names the repository as (provider, path).
 
 Checks:
     --check starts the server on a free port and runs git against it:
@@ -68,16 +69,18 @@ Checks:
         for credentials
       - a repository the token cannot see fails with the provider's own
         message, which the relay passes on
-    With --push it also commits a 256 KiB file on a new branch, pushes it with
-    a push grant, sending the pack chunked as git does for a large push, and
-    then deletes the branch. This writes to the repository, so point it at a
-    test repository.
+    With --push it also commits a 256 KiB file, pushes it with a push grant for
+    a new branch, sending the pack chunked as git does for a large push, and
+    checks that the grant can neither push another branch nor delete its own.
+    It then deletes the branch with the provider token. This writes to the
+    repository, so point it at a test repository.
 
     git runs without the user's or the system's git config, so credential
     helpers and URL rewrites cannot hide a failure.
 """
 
 import argparse
+import base64
 import importlib.machinery
 import importlib.util
 import logging
@@ -100,10 +103,10 @@ from scm.errors import GitGrantKeyError
 from scm.git.grant import generate_private_key, public_key_of, sign_grant
 from scm.git.relay import GitRelay
 from scm.git.server import GitServer
-from scm.git.types import GRANT_HEADER, GetToken, GitOperation, Grant, Readable
+from scm.git.types import GRANT_HEADER, FetchRule, GetToken, Grant, PushRule, Readable, Rule
 from scm.providers.github.provider import GitHubProvider
 from scm.providers.gitlab.provider import GitLabProvider
-from scm.types import Provider, ProviderName, Repository, RepositoryId
+from scm.types import GetGitRemoteProtocol, Provider, ProviderName, Repository, RepositoryId
 
 logger = logging.getLogger(__name__)
 
@@ -278,17 +281,20 @@ def make_git_server(
     )
 
 
-def grant_for(private_key: Secret, operation: GitOperation, repository_id: RepositoryId) -> str:
+def grant_for(private_key: Secret, rule: Rule) -> str:
     grant = Grant(
         organization_id=ORGANIZATION_ID,
         referrer=REFERRER,
-        operation=operation,
-        repositories=[repository_id],
+        rules=[rule],
         expires_at=int(time.time()) + GRANT_TTL,
     )
     return sign_grant(private_key, grant)
 
 
+def fetch_grant_for(private_key: Secret, repository_id: RepositoryId) -> str:
+    return grant_for(private_key, FetchRule(repository=repository_id, any_object=True))
+
+
 class GitHandler(BaseHTTPRequestHandler):
     """Serves /<repository path>.git/<route>, as Sentry's endpoint serves a repository's git routes.
 
@@ -408,8 +414,8 @@ def serve(backend: Backend, args: argparse.Namespace) -> None:
         (args.host, args.port), make_git_server(backend, args.public_key, args.ca_bundle), backend.provider_name
     )
     url = f"{server.url}/{backend.repository['name']}.git"
-    fetch_grant = grant_for(args.private_key, "fetch", backend.repository_id)
-    push_grant = grant_for(args.private_key, "push", backend.repository_id)
+    fetch_grant = fetch_grant_for(args.private_key, backend.repository_id)
+    push_grant = grant_for(args.private_key, PushRule(repository=backend.repository_id, branches=["my-branch"]))
     logger.info(
         f"Serving {backend.provider_name} repository {backend.repository['name']} at {url}\n\n"
         f"Grants, valid for {GRANT_TTL // 60} minutes:\n"
@@ -474,11 +480,17 @@ class Checks:
         # git shows the text of a refused first request as "remote: <message>".
         return self.check(name, result, result.returncode != 0 and f"remote: {message}" in result.stderr)
 
+    def expect_rejection(self, name: str, result: subprocess.CompletedProcess[str], reason: str) -> bool:
+        # git shows a ref the server's status report refuses as "! [remote rejected] <ref> (<reason>)".
+        rejected = "[remote rejected]" in result.stderr and f"({reason})" in result.stderr
+        return self.check(name, result, result.returncode != 0 and rejected)
+
     def run(self) -> int:
         repository_id = self.backend.repository_id
         other_repository_id = (self.backend.provider_name, f"{self.backend.repository['name']}-other")
-        fetch_grant = grant_for(self.args.private_key, "fetch", repository_id)
-        push_grant = grant_for(self.args.private_key, "push", repository_id)
+        branch = f"scm-git-server-check/{int(time.time())}"
+        fetch_grant = fetch_grant_for(self.args.private_key, repository_id)
+        push_grant = grant_for(self.args.private_key, PushRule(repository=repository_id, branches=[branch]))
 
         server = self.start()
         url = f"{server.url}/{self.backend.repository['name']}.git"
@@ -490,16 +502,15 @@ class Checks:
         cloned = self.expect_success(
             "clone with a fetch grant", self.git(fetch_grant, "clone", "--depth=1", url, str(clone))
         )
-        branch = f"scm-git-server-check/{int(time.time())}"
         self.expect_refusal(
             "push with a fetch grant is refused",
             # Refused before the provider sees it, so any local repository will do.
             self.git(fetch_grant, "push", url, f"HEAD:refs/heads/{branch}", cwd=self.local_repository()),
-            "The git grant allows fetch, not push.",
+            "The git grant allows fetch on this repository, not push.",
         )
         self.expect_refusal(
             "a grant for another repository is refused",
-            self.git(grant_for(self.args.private_key, "fetch", other_repository_id), "ls-remote", url),
+            self.git(fetch_grant_for(self.args.private_key, other_repository_id), "ls-remote", url),
             "The git grant does not cover this repository.",
         )
         server.shutdown()
@@ -519,7 +530,7 @@ class Checks:
         missing = self.backend.repository_at(f"{self.backend.repository['name']}-does-not-exist")
         server = self.start(repository=missing)
         result = self.git(
-            grant_for(self.args.private_key, "fetch", repository_id_of(self.backend.provider_name, missing)),
+            fetch_grant_for(self.args.private_key, repository_id_of(self.backend.provider_name, missing)),
             "ls-remote",
             f"{server.url}/{missing['name']}.git",
         )
@@ -557,17 +568,46 @@ class Checks:
         (clone / "scm-git-server-check.bin").write_bytes(secrets.token_bytes(256 * 1024))
         self.git(push_grant, "add", "scm-git-server-check.bin", cwd=clone).check_returncode()
         self.git(push_grant, "commit", "-m", "scm-platform git-server check", cwd=clone).check_returncode()
+        self.expect_rejection(
+            "chunked push to a branch the grant does not list is rejected",
+            self.git(
+                push_grant, "-c", "http.postBuffer=65536", "push", url, f"HEAD:refs/heads/{branch}-unlisted", cwd=clone
+            ),
+            "the git grant does not cover this branch",
+        )
         pushed = self.expect_success(
             "chunked push with a push grant",
             self.git(push_grant, "-c", "http.postBuffer=65536", "push", url, f"HEAD:refs/heads/{branch}", cwd=clone),
         )
         if pushed:
-            self.expect_success(
-                f"delete the pushed branch {branch}",
+            self.expect_rejection(
+                "deleting the pushed branch with the push grant is rejected",
                 self.git(push_grant, "push", url, f":refs/heads/{branch}", cwd=clone),
+                "the git grant does not allow deleting a branch",
+            )
+            self.expect_success(
+                f"delete the pushed branch {branch} with the token", self.delete_directly(branch, clone)
             )
         server.shutdown()
 
+    def delete_directly(self, branch: str, clone: Path) -> subprocess.CompletedProcess[str]:
+        """Delete ``branch`` on the provider with the token Sentry would use, since a push grant cannot."""
+        provider = self.backend.fetch_provider(ORGANIZATION_ID, self.backend.repository)
+        assert isinstance(provider, GetGitRemoteProtocol)
+        remote = provider.get_git_remote()
+        token = self.backend.get_token(self.backend.repository, "push")
+        assert token is not None
+        credentials = base64.b64encode(f"{remote['basic_auth_username']}:{token.reveal()}".encode()).decode()
+        # Through the environment rather than the command line, which other processes can read.
+        env = {
+            **self.env,
+            "GIT_CONFIG_COUNT": "1",
+            "GIT_CONFIG_KEY_0": "http.extraHeader",
+            "GIT_CONFIG_VALUE_0": f"Authorization: Basic {credentials}",
+        }
+        command = ["git", "-c", "credential.helper=", "push", remote["url"], f":refs/heads/{branch}"]
+        return subprocess.run(command, cwd=clone, env=env, capture_output=True, text=True, timeout=300)
+
 
 def _indent(text: str) -> str:
     return "".join(f"     {line}\n" for line in text.splitlines())
diff --git a/src/scm/errors.py b/src/scm/errors.py
index bfd674a9..71df8e7d 100644
--- a/src/scm/errors.py
+++ b/src/scm/errors.py
@@ -3,6 +3,7 @@
 type ErrorCode = Literal[
     "git_grant_expired",
     "git_invalid_grant",
+    "git_malformed_request_body",
     "git_operation_not_granted",
     "git_provider_not_supported",
     "git_remote_not_https",
@@ -51,6 +52,7 @@
 ERROR_CODES: dict[ErrorCode, str] = {
     "git_grant_expired": "The git grant has expired.",
     "git_invalid_grant": "The git grant was missing, malformed or forged.",
+    "git_malformed_request_body": "The request body is not one the git endpoint can read.",
     "git_operation_not_granted": "The git grant does not allow this operation.",
     "git_provider_not_supported": "The repository's provider has no git remote.",
     "git_remote_not_https": "The repository's git remote is not an https URL.",
@@ -170,6 +172,10 @@ class GitInvalidGrant(GitSCMCodedError):
     code = "git_invalid_grant"
 
 
+class GitMalformedRequestBody(GitSCMCodedError):
+    code = "git_malformed_request_body"
+
+
 class GitOperationNotGranted(GitSCMCodedError):
     code = "git_operation_not_granted"
 
diff --git a/src/scm/git/grant.py b/src/scm/git/grant.py
index 7763458b..07469325 100644
--- a/src/scm/git/grant.py
+++ b/src/scm/git/grant.py
@@ -44,7 +44,7 @@
 # The only algorithm the verifier accepts, with parsed Ed25519 keys. A token that names HS256 and uses a public key,
 # which is no secret, as its HMAC key is refused, as is one that names ``none``.
 _ALGORITHM = "EdDSA"
-# Claims the JWT layer owns. Every other claim must be a Grant field.
+# Claims the JWT layer owns. Every other claim must be a field of the grant.
 _JWT_CLAIMS = frozenset({"aud", "exp", "iat"})
 
 
@@ -57,6 +57,7 @@ def sign_grant(private_key: Secret, grant: Grant, *, now: float | None = None) -
     """
     key = _load_private_key(private_key)
     claims: dict[str, Any] = msgspec.to_builtins(grant)
+    claims["version"] = grant.version
     claims["exp"] = claims.pop("expires_at")
     claims["iat"] = int(time.time() if now is None else now)
     claims["aud"] = GRANT_AUDIENCE
@@ -155,8 +156,16 @@ def verify_grant(public_keys: Sequence[str], token: str, *, now: float | None =
         raise GitInvalidGrant(detail=f"The git grant is invalid: {e}") from None
 
     fields = {name: value for name, value in claims.items() if name not in _JWT_CLAIMS}
+    # A grant from before there was a second version carries no version claim.
+    version = fields.pop("version", 1)
     try:
+        if type(version) is not int or version not in (1, 2):
+            raise msgspec.ValidationError(f"Unknown version {version!r}")
         grant = msgspec.convert({**fields, "expires_at": claims["exp"]}, type=Grant, strict=True)
+        if grant.version != version:
+            raise msgspec.ValidationError(
+                f"A version {version} grant cannot hold the fields of version {grant.version}"
+            )
     except msgspec.ValidationError as e:
         # The signature matched, so the holder of the private key wrote these claims; telling them why they are
         # invalid gives nothing away.
@@ -164,7 +173,7 @@ def verify_grant(public_keys: Sequence[str], token: str, *, now: float | None =
 
     if grant.expires_at <= now:
         raise GitGrantExpired(detail="The git grant has expired.")
-    if not grant.repositories:
+    if grant.repositories is not None and not grant.repositories:
         raise GitInvalidGrant(detail="The git grant names no repositories.")
     return grant
 
diff --git a/src/scm/git/push.py b/src/scm/git/push.py
new file mode 100644
index 00000000..3f16b443
--- /dev/null
+++ b/src/scm/git/push.py
@@ -0,0 +1,249 @@
+"""Reading a push's ref updates, so GitServer can keep a push to its rule's branches.
+
+A push's request body lists the ref updates as pkt-lines, then holds the pack. GitServer reads the updates before it
+looks anything up, refuses the push unless the rule allows every update, and otherwise relays the updates it read
+followed by the rest of the body, which it never reads. gitprotocol-pack(5) describes the format, which
+receive-pack speaks in protocol versions 0 and 1 alike, the only ones it has.
+"""
+
+import re
+from collections.abc import Mapping, Sequence
+from typing import NamedTuple
+
+from scm.errors import ClientClosedRequest, GitMalformedRequestBody, GitOperationNotGranted, RpcMalformedRequestHeaders
+from scm.git.types import PushRule, Readable
+
+MAX_UPDATES_LENGTH = 64 * 1024
+"""The most bytes a push may send before its pack. An update takes under 100 bytes plus its ref's name, and a push
+from a shallow clone also lists the clone's shallow commits, about 50 bytes each."""
+
+_FLUSH = b"0000"
+# A pkt-line's length, in 4 hex digits, counts the 4 digits too.
+_PKT_LENGTH = re.compile(rb"[0-9a-fA-F]{4}")
+_MAX_PKT_LENGTH = 65520
+_OBJECT_ID = re.compile(rb"[0-9a-f]{40}|[0-9a-f]{64}")
+
+
+class RefUpdate(NamedTuple):
+    old_id: bytes
+    new_id: bytes
+    ref: bytes
+
+    @property
+    def creates(self) -> bool:
+        return not self.old_id.strip(b"0")
+
+    @property
+    def deletes(self) -> bool:
+        return not self.new_id.strip(b"0")
+
+
+class PushRequest(NamedTuple):
+    updates: list[RefUpdate]
+    capabilities: frozenset[str]
+    """What the client asked for, from its first update."""
+    head: bytes
+    """The bytes read from the body, to send on before the rest of it."""
+
+
+def read_push_request(headers: Mapping[str, str], body: Readable) -> PushRequest:
+    """Read a push's ref updates from ``body``, leaving the pack unread.
+
+    Raises GitMalformedRequestBody for a body that does not start with a list of ref updates, GitOperationNotGranted
+    for a compressed body or a signed push, whose updates come inside a certificate, and ClientClosedRequest if
+    reading the body fails.
+    """
+    encoding = next((value for name, value in headers.items() if name.lower() == "content-encoding"), "identity")
+    if encoding.strip().lower() not in ("", "identity"):
+        # git compresses fetches but never pushes.
+        raise GitOperationNotGranted(detail="The git endpoint does not read a compressed push.")
+    reader = _PktReader(body)
+    updates: list[RefUpdate] = []
+    capabilities: frozenset[str] = frozenset()
+    while (line := reader.read_line()) is not None:
+        if not updates and line.startswith(b"shallow "):
+            # A push from a shallow clone lists its shallow commits first, and they pass on unchanged.
+            if not _OBJECT_ID.fullmatch(line.removeprefix(b"shallow ")):
+                raise GitMalformedRequestBody(detail="The push lists a shallow commit that is not an object id.")
+            continue
+        if not updates and line.startswith(b"push-cert"):
+            raise GitOperationNotGranted(detail="The git endpoint does not relay signed pushes.")
+        if not updates:
+            line, _, requested = line.partition(b"\0")
+            capabilities = _capabilities(requested)
+        updates.append(_ref_update(line))
+    return PushRequest(updates=updates, capabilities=capabilities, head=reader.read_so_far())
+
+
+def refused_updates(rules: Sequence[PushRule], request: PushRequest) -> dict[bytes, str]:
+    """Return why ``rules`` refuse each ref of ``request``, or nothing if every update has a rule that allows it.
+
+    A push is relayed whole or not at all, so once one update is refused, every other ref gets a reason too.
+    """
+    if "push-options" in request.capabilities:
+        # Push options ask the provider to act on the push, such as GitLab opening a merge request or skipping CI.
+        return {update.ref: "the git endpoint does not relay push options" for update in request.updates}
+    refused: dict[bytes, str] = {}
+    for update in request.updates:
+        refusals = [_refusal(rule, update) for rule in rules]
+        reasons = [refusal for refusal in refusals if refusal is not None]
+        if len(reasons) == len(refusals):
+            # The reason of the rule the update came closest to passing says the most.
+            refused[update.ref] = max(reasons, default=_NOT_COVERED)[1]
+    if refused:
+        for update in request.updates:
+            refused.setdefault(update.ref, "refused with the rest of the push")
+    return refused
+
+
+_NOT_A_BRANCH = (0, "the git grant allows pushing branches only")
+_NOT_COVERED = (1, "the git grant does not cover this branch")
+_NOT_NEW = (2, "the git grant allows only creating this branch")
+_DELETES = (3, "the git grant does not allow deleting a branch")
+
+
+def _refusal(rule: PushRule, update: RefUpdate) -> tuple[int, str] | None:
+    """Return why ``rule`` refuses ``update``, ranked by how far the update got, or None if the rule allows it."""
+    if not update.ref.startswith(b"refs/heads/"):
+        return _NOT_A_BRANCH
+    try:
+        branch = update.ref.removeprefix(b"refs/heads/").decode()
+    except UnicodeDecodeError:
+        return _NOT_COVERED
+    if not rule.covers(branch):
+        return _NOT_COVERED
+    if update.deletes:
+        return _DELETES
+    if rule.create_only and not update.creates:
+        return _NOT_NEW
+    return None
+
+
+def refusal_report(request: PushRequest, refused: Mapping[bytes, str]) -> bytes | None:
+    """Return the response body that tells git why each ref was refused, or None if the client asked for none.
+
+    git shows each ref as ``! [remote rejected] <branch> (<reason>)``, as for a provider's own refusal. A refused
+    HTTP request would show only its status code.
+    """
+    if not {"report-status", "report-status-v2"} & request.capabilities:
+        return None
+    # "unpack ok" makes git show each ref's reason rather than an unpack failure. Each "ng" line is shorter than
+    # the update it answers, so it fits in a pkt-line.
+    report = (
+        _pkt(b"unpack ok\n")
+        + b"".join(_pkt(b"ng %s %s\n" % (update.ref, refused[update.ref].encode())) for update in request.updates)
+        + _FLUSH
+    )
+    if "side-band-64k" in request.capabilities:
+        band_size = _MAX_PKT_LENGTH - 5
+    elif "side-band" in request.capabilities:
+        band_size = 1000 - 5
+    else:
+        return report
+    # On a side band the report travels on band 1, and the response ends with a flush of its own.
+    return b"".join(_pkt(b"\1" + report[i : i + band_size]) for i in range(0, len(report), band_size)) + _FLUSH
+
+
+def bounded_body(headers: Mapping[str, str], body: Readable) -> Readable:
+    """Return ``body``, ending after its Content-Length if it has one, as the relay reads it.
+
+    A caller may pass a stream that goes on past the body, such as a connection's socket, and reading the pushed
+    updates or a refused pack past the body's end would wait for bytes that never come.
+    """
+    raw_length = next((value for name, value in headers.items() if name.lower() == "content-length"), None)
+    if raw_length is None:
+        return body
+    try:
+        length = int(raw_length)
+    except ValueError:
+        raise RpcMalformedRequestHeaders(detail="Could not parse Content-Length header") from None
+    if length < 0:
+        raise RpcMalformedRequestHeaders(detail="Could not parse Content-Length header")
+    return _BoundedBody(body, length)
+
+
+class _BoundedBody:
+    def __init__(self, body: Readable, length: int) -> None:
+        self.body = body
+        self.remaining = length
+
+    def read(self, size: int = -1, /) -> bytes:
+        size = self.remaining if size < 0 else min(size, self.remaining)
+        if size == 0:
+            return b""
+        data = self.body.read(size)
+        self.remaining -= len(data)
+        return data
+
+
+class ReplayedBody:
+    """A request body whose first bytes were read already, sending them before the rest."""
+
+    def __init__(self, head: bytes, rest: Readable) -> None:
+        self.head = head
+        self.rest = rest
+
+    def read(self, size: int = -1, /) -> bytes:
+        if not self.head:
+            return self.rest.read(size)
+        if size < 0:
+            data, self.head = self.head + self.rest.read(), b""
+            return data
+        data, self.head = self.head[:size], self.head[size:]
+        return data
+
+
+class _PktReader:
+    def __init__(self, body: Readable) -> None:
+        self.body = body
+        self.read_bytes = bytearray()
+
+    def read_line(self) -> bytes | None:
+        """Return the next pkt-line's payload without its trailing newline, or None for a flush."""
+        header = self._read(4)
+        if not _PKT_LENGTH.fullmatch(header):
+            raise GitMalformedRequestBody(detail="The push's ref updates are not pkt-lines.")
+        length = int(header, 16)
+        if length == 0:
+            return None
+        # 1 to 3 are protocol v2's special packets, which a push never sends.
+        if length < 4 or length > _MAX_PKT_LENGTH:
+            raise GitMalformedRequestBody(detail="The push's ref updates are not pkt-lines.")
+        return self._read(length - 4).removesuffix(b"\n")
+
+    def read_so_far(self) -> bytes:
+        return bytes(self.read_bytes)
+
+    def _read(self, size: int) -> bytes:
+        if len(self.read_bytes) + size > MAX_UPDATES_LENGTH:
+            raise GitMalformedRequestBody(detail=f"The push's ref updates are longer than {MAX_UPDATES_LENGTH} bytes.")
+        data = bytearray()
+        while len(data) < size:
+            try:
+                chunk = self.body.read(size - len(data))
+            except OSError:
+                raise ClientClosedRequest(detail="The client's connection failed while it sent its push.") from None
+            if not chunk:
+                raise GitMalformedRequestBody(detail="The push ended before its ref updates did.")
+            data += chunk
+        self.read_bytes += data
+        return bytes(data)
+
+
+def _capabilities(requested: bytes) -> frozenset[str]:
+    try:
+        return frozenset(requested.decode("ascii").split())
+    except UnicodeDecodeError:
+        raise GitMalformedRequestBody(detail="The push's capabilities are not ASCII.") from None
+
+
+def _ref_update(line: bytes) -> RefUpdate:
+    old_id, _, rest = line.partition(b" ")
+    new_id, _, ref = rest.partition(b" ")
+    if not (_OBJECT_ID.fullmatch(old_id) and _OBJECT_ID.fullmatch(new_id) and len(old_id) == len(new_id) and ref):
+        raise GitMalformedRequestBody(detail="The push's ref updates are malformed.")
+    return RefUpdate(old_id=old_id, new_id=new_id, ref=ref)
+
+
+def _pkt(payload: bytes) -> bytes:
+    return b"%04x" % (len(payload) + 4) + payload
diff --git a/src/scm/git/relay.py b/src/scm/git/relay.py
index b65848ae..fbed82bd 100644
--- a/src/scm/git/relay.py
+++ b/src/scm/git/relay.py
@@ -20,7 +20,7 @@
     RpcMalformedRequestHeaders,
     SCMCodedError,
 )
-from scm.git.types import GitOperation, Grant, Readable
+from scm.git.types import FetchRule, GitOperation, Grant, Readable
 from scm.rpc.types import StreamResponse
 from scm.secret import Secret
 from scm.types import GitRemote, RepositoryId
@@ -67,17 +67,36 @@
 def authorize_request(grant: Grant, repository_id: RepositoryId, method: str, route: str, query: str) -> GitOperation:
     """Return the git operation a request belongs to, refusing the request unless the grant allows it.
 
-    A fetch grant reaches only upload-pack and a push grant only receive-pack. The repository must be listed in
-    the grant in the same form the request uses: a database id does not match the same repository's
-    ``(provider, external id)`` pair.
+    Fetches reach only upload-pack and pushes only receive-pack. The grant must name the repository in the same form
+    the request uses: a database id does not match the same repository's ``(provider, external id)`` pair.
+
+    A grant with rules allows a request only if it has a rule for the request's repository and operation. Push rules'
+    branches are kept by GitServer, which reads the push's ref updates before it relays the push, so a caller that
+    relays a push some other way must check them itself. A fetch needs a fetch rule for any object, since nothing
+    reads a fetch's body to keep it to a rule's branches.
     """
     operation = _ROUTES.get((method, route, query))
     if operation is None:
         raise GitRouteNotSupported(detail=f"{method} {route} is not a git fetch or push route.")
-    if operation != grant.operation:
-        raise GitOperationNotGranted(detail=f"The git grant allows {grant.operation}, not {operation}.")
-    if repository_id not in grant.repositories:
-        raise GitRepositoryNotGranted(detail="The git grant does not cover this repository.")
+    if grant.rules is None:
+        # The version 1 form, one operation on every listed repository.
+        if operation != grant.operation:
+            raise GitOperationNotGranted(detail=f"The git grant allows {grant.operation}, not {operation}.")
+        if repository_id not in (grant.repositories or ()):
+            raise GitRepositoryNotGranted(detail="The git grant does not cover this repository.")
+        return operation
+
+    rules = grant.rules_for(repository_id, operation)
+    if not rules:
+        other: GitOperation = "push" if operation == "fetch" else "fetch"
+        if not grant.rules_for(repository_id, other):
+            raise GitRepositoryNotGranted(detail="The git grant does not cover this repository.")
+        raise GitOperationNotGranted(detail=f"The git grant allows {other} on this repository, not {operation}.")
+    # Delete this once GitServer keeps a fetch to its rule's branches.
+    if operation == "fetch" and not any(isinstance(rule, FetchRule) and rule.any_object for rule in rules):
+        raise GitOperationNotGranted(
+            detail="The git endpoint cannot keep a fetch to a rule's branches yet, so it refuses the rule."
+        )
     return operation
 
 
diff --git a/src/scm/git/server.py b/src/scm/git/server.py
index e49bedf1..4548a1b7 100644
--- a/src/scm/git/server.py
+++ b/src/scm/git/server.py
@@ -1,10 +1,26 @@
+import contextlib
 from collections.abc import Callable, Mapping, Sequence
 from urllib.parse import urlsplit
 
-from scm.errors import ErrorCode, GitProviderNotSupported, GitTokenRejected, GitTokenUnavailable, SCMCodedError
+from scm.errors import (
+    ErrorCode,
+    GitOperationNotGranted,
+    GitProviderNotSupported,
+    GitTokenRejected,
+    GitTokenUnavailable,
+    SCMCodedError,
+)
 from scm.git.grant import require_public_keys, verify_grant
-from scm.git.relay import GitRelay, ResponseBody, authorize_request
-from scm.git.types import GRANT_HEADER, GetToken, Readable
+from scm.git.push import (
+    PushRequest,
+    ReplayedBody,
+    bounded_body,
+    read_push_request,
+    refusal_report,
+    refused_updates,
+)
+from scm.git.relay import DEFAULT_CHUNK_SIZE, GitRelay, ResponseBody, authorize_request
+from scm.git.types import GRANT_HEADER, GetToken, PushRule, Readable
 from scm.helpers import initialize_provider
 from scm.rpc.types import StreamResponse
 from scm.secret import Secret
@@ -15,6 +31,7 @@
 _STATUS: dict[ErrorCode, int] = {
     # The client's own request is malformed, or is not a git request this endpoint serves.
     "rpc_malformed_request_headers": 400,
+    "git_malformed_request_body": 400,
     "git_route_not_supported": 404,
     # The grant does not allow the request.
     "git_invalid_grant": 403,
@@ -58,6 +75,13 @@ class GitServer:
     does not allow never reaches the database, the token supplier or the provider. Give it a relay whose adapter
     refuses internal addresses; see ``GitRelay``.
 
+    A push under a grant with rules is kept to its push rules. The server reads the push's ref updates, which come
+    before the pack, and refuses the whole push if no rule allows one of them: a branch no rule covers, an existing
+    branch under a ``create_only`` rule, a deletion, or any ref that isn't a branch, such as a tag. It refuses push
+    options and signed pushes too. A client that asked for a status report gets one, a 200 that git shows as a
+    rejected ref with its reason. A push the rules allow is relayed with the updates the server read followed by the
+    rest of the body, unread. A push under a version 1 grant is relayed unread, as before rules.
+
     The repository's provider says where its git endpoint is, through ``GetGitRemoteProtocol``. A repository whose
     provider has no git remote is refused.
 
@@ -129,12 +153,27 @@ def handle(
         counted_body = _CountingReader(body) if body is not None else None
         context = {"repository_id": str(repository_id)}
         referrer_tag: dict[str, str] = {}
+        push_body: Readable | None = None
+        push: PushRequest | None = None
+        refused: dict[bytes, str] = {}
         try:
             grant = verify_grant(self.public_keys, grant_token.reveal())
             context["organization_id"] = str(grant.organization_id)
             referrer_tag["referrer"] = context["referrer"] = grant.referrer
             operation = authorize_request(grant, repository_id, method, route, query)
             context["operation"] = operation
+            relayed_body: Readable | None = counted_body
+            push_rules = [rule for rule in grant.rules_for(repository_id, operation) if isinstance(rule, PushRule)]
+            if push_rules and method == "POST" and counted_body is not None:
+                push_body = bounded_body(headers, counted_body)
+                push = read_push_request(headers, push_body)
+                refused = refused_updates(push_rules, push)
+                if refused:
+                    ref, reason = next(iter(refused.items()))
+                    raise GitOperationNotGranted(
+                        detail=f"The git grant does not allow this push: {ref.decode(errors='replace')} ({reason})."
+                    )
+                relayed_body = ReplayedBody(push.head, push_body)
             provider = initialize_provider(
                 grant.organization_id,
                 repository_id,
@@ -163,7 +202,7 @@ def handle(
                     "read it."
                 )
             response = self.relay.forward(
-                remote, token, method=method, route=route, query=query, headers=headers, body=counted_body
+                remote, token, method=method, route=route, query=query, headers=headers, body=relayed_body
             )
             if response.status_code == 401:
                 # The provider rejected the token Sentry supplied. Passed on, the 401 would make git ask the client
@@ -174,6 +213,11 @@ def handle(
             self.record_count("sentry.scm.git.request.failed", 1, {"code": e.code} | referrer_tag)
             if e.code != "client_closed_request":
                 self.emit_error(e, context)
+            if push is not None and refused and push_body is not None:
+                _drain(push_body)
+                status_report = refusal_report(push, refused)
+                if status_report is not None:
+                    return _report_response(status_report)
             return _error_response(e)
         except Exception:
             self.record_count("sentry.scm.git.request.failed", 1, {"code": "unhandled_exception"} | referrer_tag)
@@ -279,6 +323,26 @@ def _end(self, failure: str | None) -> None:
             self.finish(failure, self.bytes_sent)
 
 
+def _drain(body: Readable) -> None:
+    # git reads the response only once it has sent the whole push, and a connection closed before then shows only as
+    # a broken pipe, hiding the reason, so a refused push's pack is read to the end and dropped.
+    with contextlib.suppress(OSError, SCMCodedError):
+        while body.read(DEFAULT_CHUNK_SIZE):
+            pass
+
+
+def _report_response(report: bytes) -> StreamResponse:
+    return StreamResponse(
+        status_code=200,
+        headers={
+            "Content-Type": "application/x-git-receive-pack-result",
+            "Content-Length": str(len(report)),
+            "Cache-Control": "no-cache",
+        },
+        content=iter([report]),
+    )
+
+
 def _error_response(error: SCMCodedError) -> StreamResponse:
     # A code missing from the map comes from a failure nobody planned for, which is the server's fault.
     status = _STATUS.get(error.code, 500)
diff --git a/src/scm/git/types.py b/src/scm/git/types.py
index 66239d0d..45e1e853 100644
--- a/src/scm/git/types.py
+++ b/src/scm/git/types.py
@@ -1,15 +1,18 @@
+import re
 from collections.abc import Callable
-from typing import Literal, Protocol
+from typing import ClassVar, Literal, Protocol
 
 import msgspec
 
 from scm.secret import Secret
-from scm.types import Referrer, Repository, RepositoryId
+from scm.types import BranchName, Referrer, Repository, RepositoryId
 
 type GitOperation = Literal["fetch", "push"]
-"""What a grant lets its holder do. ``scm.git.relay`` maps each to its git routes."""
+"""What a grant or rule lets its holder do. ``scm.git.relay`` maps each to its git routes."""
 
 GRANT_VERSION: Literal[1] = 1
+"""Deprecated: the version of a grant without rules, which ``Grant.version`` gives. Remove it together with
+``Grant.operation`` and ``Grant.repositories``."""
 
 GRANT_HEADER = "X-Scm-Git-Grant"
 """The request header that carries a signed grant to the git endpoint."""
@@ -17,23 +20,119 @@
 GRANT_MAX_LENGTH = 4096
 """The longest signed grant, in characters. The grant travels in a request header, and sandbox-router's
 ``PUT /grant/{sandbox_id}`` refuses a longer one, so a change here must change the router too. A grant grows with
-its repositories: a grant for one is under 450 characters, and the claims are base64-encoded, so each repository
-named as ``(provider, path)`` adds about 4/3 of its path's length plus 20, e.g. 52 for a 25-character path. The
-limit is a conservative pick rather than a measured ceiling: header limits further along the path are typically
-around 8 KiB."""
-
-
-class Grant(msgspec.Struct, frozen=True, kw_only=True, forbid_unknown_fields=True):
-    """Permission to run one git operation against a fixed set of an organization's repositories.
+its rules: a grant with one is under 500 characters, and the claims are base64-encoded, so a fetch rule for any
+object of a repository named as ``(provider, path)`` adds about 4/3 of its path's length plus 87, e.g. 120 for a
+25-character path, and each branch a rule lists adds about 4/3 of its name's length plus 4. The limit is a
+conservative pick rather than a measured ceiling: header limits further along the path are typically around
+8 KiB."""
+
+
+class FetchRule(
+    msgspec.Struct,
+    frozen=True,
+    kw_only=True,
+    forbid_unknown_fields=True,
+    omit_defaults=True,
+    tag="fetch",
+    tag_field="operation",
+):
+    """Lets the holder fetch from ``repository``: any of its objects, or only what its listed ``branches`` reach.
+
+    A rule sets exactly one of the two, so a rule that leaves its branches out is refused rather than read as one
+    that reads everything. ``any_object`` reaches every object the provider serves, including objects no branch
+    reaches and, on GitHub, objects of the repository's forks, so give it only to code that runs no agent, such as a
+    checkout before the agent starts.
+    """
 
-    Seer signs a grant per coding job and Sentry's git endpoint verifies it on every request. Each repository is
-    a ``RepositoryId`` under the grant's organization, as in the RPC endpoint: Sentry's repository id or a
-    ``(provider, external id)`` pair. A fetch grant holds one repository today. The list leaves room for
-    submodules and git-based dependencies, but only from repositories Sentry has connected for the organization:
-    naming any other repository, such as a public dependency, would need a new kind of entry and a new version.
+    operation: ClassVar[GitOperation] = "fetch"
+    repository: RepositoryId
+    branches: list[BranchName] | None = None
+    """Branch names as ``git branch`` takes them, without ``refs/heads/``."""
+    any_object: bool = False
+
+    def __post_init__(self) -> None:
+        if self.branches is None and not self.any_object:
+            raise ValueError("A fetch rule must list its branches or set any_object.")
+        if self.branches is not None and self.any_object:
+            raise ValueError("A fetch rule must not both list branches and set any_object.")
+        if self.branches is not None:
+            _check_branches(self.branches)
+
+
+class PushRule(
+    msgspec.Struct,
+    frozen=True,
+    kw_only=True,
+    forbid_unknown_fields=True,
+    omit_defaults=True,
+    tag="push",
+    tag_field="operation",
+):
+    """Lets the holder create or update the branches of ``repository`` the rule covers, and no other ref: no tag,
+    and no deletion.
+
+    The rule covers the branches it lists in ``branches``, or every branch whose whole name matches one of
+    ``branch_patterns``, and sets exactly one of the two. With ``create_only`` it allows only creating a covered
+    branch that doesn't exist yet, so a pattern can let the holder start new branches without moving any that exist.
+    Without it, a push may move a covered branch to any commit, so a force push passes.
+
+    Don't cover the repository's default branch, by name or by pattern. The git endpoint doesn't know which branch
+    that is, so keeping it out is the signer's job, with the provider's branch protection behind it.
+    """
 
-    A grant with a field this version does not know is refused rather than read without it. If a later version
-    adds a field that restricts a grant, a verifier that ignored the field would allow more than the grant does.
+    operation: ClassVar[GitOperation] = "push"
+    repository: RepositoryId
+    branches: list[BranchName] | None = None
+    """Branch names as ``git branch`` takes them, without ``refs/heads/``."""
+    branch_patterns: list[str] | None = None
+    """Python regular expressions, each matched against a branch's whole name without ``refs/heads/``, such as
+    ``seer/.+``. The pushing client chooses the names they are matched against, so a pattern must not be one that
+    backtracks catastrophically."""
+    create_only: bool = False
+    """Allow only creating a covered branch. git sends a new branch with an old object id of all zeros, and the
+    provider refuses such an update if the branch already exists."""
+
+    def __post_init__(self) -> None:
+        if self.branches is None and self.branch_patterns is None:
+            raise ValueError("A push rule must list its branches or set branch_patterns.")
+        if self.branches is not None and self.branch_patterns is not None:
+            raise ValueError("A push rule must not both list branches and set branch_patterns.")
+        if self.branches is not None:
+            _check_branches(self.branches)
+        if self.branch_patterns is not None:
+            _check_patterns(self.branch_patterns)
+
+    def covers(self, branch: BranchName) -> bool:
+        """Return whether the rule covers ``branch``, a name without ``refs/heads/``."""
+        if self.branches is not None:
+            return branch in self.branches
+        return any(re.fullmatch(pattern, branch) for pattern in self.branch_patterns or ())
+
+
+type Rule = FetchRule | PushRule
+
+
+class Grant(msgspec.Struct, frozen=True, kw_only=True, forbid_unknown_fields=True, omit_defaults=True):
+    """Permission to run git operations on an organization's repositories, as a list of rules.
+
+    Seer signs a grant for each sandbox, and for its own git requests, and Sentry's git endpoint verifies it on
+    every request. A sandbox holds one grant at a time, so a grant that lets a pod both fetch and push holds a rule
+    for each. A request runs only if a rule for its repository and operation allows it. Several rules may name the
+    same repository and operation, and a request runs if any of them allows it, so their order doesn't matter: a
+    ``create_only`` rule with a pattern can sit next to a rule listing a branch the holder created and may update.
+
+    Each rule names its repository as a ``RepositoryId`` under the grant's organization, as in the RPC endpoint:
+    Sentry's repository id or a ``(provider, external id)`` pair. Several repositories leave room for submodules and
+    git-based dependencies, but only from repositories Sentry has connected for the organization: naming any other
+    repository, such as a public dependency, would need a new kind of rule.
+
+    A grant sets ``rules`` or, in the deprecated version 1 form, ``operation`` and ``repositories``, which allow one
+    operation on every listed repository. A version 1 push lets its holder update any ref, including tags and
+    deletions, since GitServer relays it unread. Verifiers read both forms, so a signer can switch once they do.
+
+    A grant or rule with a field the verifier does not know is refused rather than read without it, because a
+    verifier that ignored a field restricting the grant would allow more than the grant does. So a new field needs
+    only the verifiers to know it before the signer uses it, and a new version only when a field changes meaning.
     """
 
     organization_id: int
@@ -41,12 +140,69 @@ class Grant(msgspec.Struct, frozen=True, kw_only=True, forbid_unknown_fields=Tru
     """Who the grant is for, e.g. a coding job or Seer itself, as ``referrer`` is elsewhere in this library.
     Sentry tags the git endpoint's metrics with it. Seer sets it, because git requests carry nothing else that
     tells callers apart."""
-    operation: GitOperation
-    repositories: list[RepositoryId]
     expires_at: int
     """Unix time in seconds. Every grant expires."""
-    version: Literal[1] = GRANT_VERSION
-    """The format version. A verifier refuses a version it does not know."""
+    rules: list[Rule] | None = None
+    operation: GitOperation | None = None
+    """Deprecated, with ``repositories``: the version 1 form. Remove both once no signer sets them, and make
+    ``rules`` required."""
+    repositories: list[RepositoryId] | None = None
+    """Deprecated, with ``operation``."""
+
+    def __post_init__(self) -> None:
+        if self.rules is not None:
+            if self.operation is not None or self.repositories is not None:
+                raise ValueError("A git grant sets either rules or operation and repositories, not both.")
+            if not self.rules:
+                raise ValueError("A git grant must hold at least one rule.")
+        elif self.operation is None or self.repositories is None:
+            raise ValueError("A git grant must set rules, or both operation and repositories.")
+
+    @property
+    def version(self) -> Literal[1, 2]:
+        """The format the grant is signed in: 2 with rules, and 1 with operation and repositories."""
+        return 1 if self.rules is None else 2
+
+    def rules_for(self, repository: RepositoryId, operation: GitOperation) -> list[Rule]:
+        """Return the rules for ``operation`` on ``repository``, named as the rules name it.
+
+        A version 1 grant has no rules, so it returns none."""
+        return [rule for rule in self.rules or () if rule.repository == repository and rule.operation == operation]
+
+
+# What git check-ref-format refuses in a ref name: control characters, space, ~ ^ : ? * [ \, "..", "@{" and "//".
+_NOT_IN_A_REF = re.compile(r"[\x00-\x20\x7f~^:?*\[\\]|\.\.|@\{|//")
+
+
+def _check_patterns(patterns: list[str]) -> None:
+    if not patterns:
+        raise ValueError("A rule's branch pattern list must not be empty.")
+    for pattern in patterns:
+        if not pattern:
+            raise ValueError("A branch pattern must not be empty.")
+        try:
+            re.compile(pattern)
+        except re.error as e:
+            raise ValueError(f"{pattern!r} is not a regular expression: {e}") from None
+
+
+def _check_branches(branches: list[BranchName]) -> None:
+    if not branches:
+        raise ValueError("A rule's branch list must not be empty.")
+    for name in branches:
+        # git check-ref-format's rules for refs/heads/<name>, with what git branch refuses on top (a leading "-" and
+        # "HEAD"). A name starting with refs/ is a valid branch, but it's far more likely a full ref name by mistake.
+        if (
+            not name
+            or name in ("@", "HEAD")
+            or name.startswith(("-", "/", "refs/"))
+            or name.endswith(("/", "."))
+            or _NOT_IN_A_REF.search(name)
+            or any(part.startswith(".") or part.endswith(".lock") for part in name.split("/"))
+        ):
+            raise ValueError(
+                f"{name!r} is not a branch name. Name a branch as git branch takes it, without refs/heads/."
+            )
 
 
 type GetToken = Callable[[Repository, GitOperation], Secret | None]
diff --git a/tests/unit/git_builders.py b/tests/unit/git_builders.py
index fe1418c1..078e738e 100644
--- a/tests/unit/git_builders.py
+++ b/tests/unit/git_builders.py
@@ -7,7 +7,7 @@
 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat, load_pem_private_key
 
 from scm.git.grant import generate_private_key
-from scm.git.types import Grant
+from scm.git.types import FetchRule, Grant
 from scm.providers.cursor_origin.provider import CursorOriginProvider
 from scm.providers.github.provider import GitHubProvider
 from scm.providers.gitlab.provider import GitLabProvider
@@ -48,7 +48,19 @@ def make_provider(organization_id: int, repository: Repository) -> Provider | No
 
 
 def make_grant(**overrides) -> Grant:
-    """A fetch grant for the default repository, named both ways a grant can name it."""
+    """A grant to fetch any object of the default repository, named both ways a rule can name it."""
+    fields = {
+        "organization_id": 1,
+        "referrer": "seer.coding_job",
+        "rules": [FetchRule(repository=5, any_object=True), FetchRule(repository=("github", "123"), any_object=True)],
+        "expires_at": FAR_FUTURE,
+    }
+    fields.update(overrides)
+    return Grant(**fields)  # type: ignore[arg-type]
+
+
+def make_v1_grant(**overrides) -> Grant:
+    """A version 1 fetch grant for the default repository, named both ways a grant can name it."""
     fields = {
         "organization_id": 1,
         "referrer": "seer.coding_job",
@@ -80,3 +92,30 @@ def make_key_pair() -> KeyPair:
         # Serialized here rather than by public_key_of, so the tests check public_key_of against cryptography.
         public=key.public_key().public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo).decode(),
     )
+
+
+ZERO_ID = "0" * 40
+OLD_ID = "1" * 40
+NEW_ID = "2" * 40
+PACK = b"PACK\x00\x00\x00\x02" + b"pack contents" * 16
+
+
+def pkt(payload: bytes) -> bytes:
+    return b"%04x" % (len(payload) + 4) + payload
+
+
+def make_push_body(
+    *updates: tuple[str, str, str],
+    capabilities: str = "report-status side-band-64k quiet object-format=sha1 agent=git/2.54.0",
+    shallow: tuple[str, ...] = (),
+    pack: bytes = PACK,
+) -> bytes:
+    """A receive-pack request body as git sends it: shallow commits, ref updates (old id, new id, ref) and the pack.
+
+    git ends no update with a newline, and puts its capabilities after a NUL on the first.
+    """
+    lines = [pkt(f"shallow {commit}\n".encode()) for commit in shallow]
+    for index, (old_id, new_id, ref) in enumerate(updates):
+        line = f"{old_id} {new_id} {ref}".encode()
+        lines.append(pkt(line + b"\0 " + capabilities.encode() if index == 0 else line))
+    return b"".join(lines) + b"0000" + pack
diff --git a/tests/unit/test_git_grant.py b/tests/unit/test_git_grant.py
index a8d07ff8..1b5db7b3 100644
--- a/tests/unit/test_git_grant.py
+++ b/tests/unit/test_git_grant.py
@@ -31,10 +31,10 @@
     sign_grant,
     verify_grant,
 )
-from scm.git.types import GRANT_MAX_LENGTH, Grant
+from scm.git.types import GRANT_MAX_LENGTH, FetchRule, Grant, PushRule
 from scm.secret import Secret
 from tests.unit.error_reports import assert_not_in_frames
-from tests.unit.git_builders import FAR_FUTURE, KeyPair, make_grant, make_key_pair
+from tests.unit.git_builders import FAR_FUTURE, KeyPair, make_grant, make_key_pair, make_v1_grant
 
 # Before the real time: PyJWT checks iat against the real clock, so a grant must not be issued after it.
 NOW = 1_700_000_000
@@ -51,7 +51,7 @@ def expected_key_id(key: KeyPair) -> str:
 
 
 def claims_of(grant: Grant) -> dict:
-    claims = msgspec.to_builtins(grant)
+    claims = msgspec.to_builtins(grant) | {"version": grant.version}
     claims["exp"] = claims.pop("expires_at")
     return claims | {"iat": NOW, "aud": GRANT_AUDIENCE}
 
@@ -117,9 +117,17 @@ def test_public_key_of_the_signing_key(self):
         # make_key_pair serializes the public key with cryptography directly, not through public_key_of.
         assert public_key_of(KEY.private) == KEY.public
 
-    def test_push_grant(self):
-        grant = make_grant(operation="push")
-        assert verify_grant([KEY.public], sign_grant(KEY.private, grant, now=NOW), now=NOW).operation == "push"
+    def test_rules_of_each_kind(self):
+        grant = make_grant(
+            rules=[
+                FetchRule(repository=5, branches=["main", "seer/fix"]),
+                PushRule(repository=5, branches=["seer/fix"]),
+                FetchRule(repository=("gitlab", "gitlab.com:7"), any_object=True),
+            ]
+        )
+        verified = verify_grant([KEY.public], sign_grant(KEY.private, grant, now=NOW), now=NOW)
+        assert verified == grant
+        assert [rule.operation for rule in verified.rules] == ["fetch", "push", "fetch"]
 
     def test_token_is_a_jwt_with_the_grants_claims(self):
         token = sign_grant(KEY.private, make_grant(), now=NOW)
@@ -127,9 +135,11 @@ def test_token_is_a_jwt_with_the_grants_claims(self):
         assert jwt.decode(token, options={"verify_signature": False}) == {
             "organization_id": 1,
             "referrer": "seer.coding_job",
-            "operation": "fetch",
-            "repositories": [5, ["github", "123"]],
-            "version": 1,
+            "rules": [
+                {"operation": "fetch", "repository": 5, "any_object": True},
+                {"operation": "fetch", "repository": ["github", "123"], "any_object": True},
+            ],
+            "version": 2,
             "exp": FAR_FUTURE,
             "iat": NOW,
             "aud": GRANT_AUDIENCE,
@@ -266,8 +276,10 @@ def test_other_asymmetric_algorithm(self):
 
 
 def grant_with_repositories(count: int) -> Grant:
-    # Names of one length, so each repository adds the same to the grant.
-    return make_grant(repositories=[("github", f"getsentry/repository-{i:04d}") for i in range(count)])
+    # Names of one length, so each rule adds the same to the grant.
+    return make_grant(
+        rules=[FetchRule(repository=("github", f"getsentry/repository-{i:04d}"), any_object=True) for i in range(count)]
+    )
 
 
 class TestLength:
@@ -291,14 +303,14 @@ def test_longest_grant_that_fits_signs(self):
             except ValueError:
                 break
             token, count = longer, count + 1
-        # One more repository adds about 50 characters.
-        assert GRANT_MAX_LENGTH - 100 < len(token) <= GRANT_MAX_LENGTH
+        # One more rule adds about 120 characters.
+        assert GRANT_MAX_LENGTH - 150 < len(token) <= GRANT_MAX_LENGTH
         assert verify_grant([KEY.public], token, now=NOW) == grant_with_repositories(count)
 
     def test_grant_for_one_repository(self):
         # GRANT_MAX_LENGTH's docstring sizes grants from this.
-        grant = make_grant(repositories=[("github", "getsentry/scm-platform")])
-        assert len(sign_grant(KEY.private, grant, now=NOW)) < 450
+        grant = make_grant(rules=[FetchRule(repository=("github", "getsentry/scm-platform"), any_object=True)])
+        assert len(sign_grant(KEY.private, grant, now=NOW)) < 500
 
 
 def other_key_types() -> list:
@@ -505,10 +517,12 @@ def test_missing(self):
             {"organization_id": True},
             {"organization_id": "1"},
             {"referrer": None},
-            {"operation": "delete"},
-            {"version": 2},
-            {"repositories": [["github"]]},
-            {"repositories": [["unknown_provider", "123"]]},
+            {"version": 1},
+            {"version": 3},
+            {"version": "2"},
+            {"rules": None},
+            {"rules": [{"operation": "fetch", "repository": ["github"], "any_object": True}]},
+            {"rules": [{"operation": "fetch", "repository": ["unknown_provider", "123"], "any_object": True}]},
             {"exp": NOW + 600.5},
             {"narrower_than_you_think": True},
         ],
@@ -522,5 +536,278 @@ def test_referrer_is_required(self):
         del claims["referrer"]
         assert_invalid(sign_claims(claims), "referrer")
 
+    def test_version_1_claims_labelled_version_2(self):
+        assert_invalid(sign_claims(claims_of(make_v1_grant()) | {"version": 2}), "payload is invalid")
+
+
+def rule_claims(*rules: dict) -> dict:
+    return claims_of(make_grant()) | {"rules": list(rules)}
+
+
+class TestRules:
+    """A grant's rules, checked both where a grant is built and where it is verified, whoever signed it."""
+
+    def test_fetch_and_push_rules_for_one_repository(self):
+        grant = make_grant(
+            rules=[FetchRule(repository=5, branches=["main"]), PushRule(repository=5, branches=["seer/fix"])]
+        )
+        assert verify_grant([KEY.public], sign_grant(KEY.private, grant, now=NOW), now=NOW) == grant
+
+    def test_claims_hold_only_what_a_rule_sets(self):
+        # Shorter, and a reader sees which fetch rules reach any object.
+        grant = make_grant(
+            rules=[
+                FetchRule(repository=5, branches=["main"]),
+                FetchRule(repository=6, any_object=True),
+                PushRule(repository=5, branches=["seer/fix"]),
+                PushRule(repository=5, branch_patterns=["seer/.+"], create_only=True),
+            ]
+        )
+        claims = jwt.decode(sign_grant(KEY.private, grant, now=NOW), options={"verify_signature": False})
+        assert claims["rules"] == [
+            {"operation": "fetch", "repository": 5, "branches": ["main"]},
+            {"operation": "fetch", "repository": 6, "any_object": True},
+            {"operation": "push", "repository": 5, "branches": ["seer/fix"]},
+            {"operation": "push", "repository": 5, "branch_patterns": ["seer/.+"], "create_only": True},
+        ]
+
+    @pytest.mark.parametrize(
+        "rule,error",
+        [
+            ({"operation": "fetch", "repository": 5}, "must list its branches or set any_object"),
+            ({"operation": "fetch", "repository": 5, "any_object": False}, "must list its branches or set any_object"),
+            ({"operation": "fetch", "repository": 5, "branches": None}, "must list its branches or set any_object"),
+            (
+                {"operation": "fetch", "repository": 5, "branches": ["main"], "any_object": True},
+                "must not both list branches and set any_object",
+            ),
+            ({"operation": "fetch", "repository": 5, "branches": []}, "must not be empty"),
+            ({"operation": "push", "repository": 5, "branches": []}, "must not be empty"),
+            ({"operation": "push", "repository": 5}, "must list its branches or set branch_patterns"),
+            ({"operation": "push", "repository": 5, "branches": None}, "must list its branches or set branch_patterns"),
+            (
+                {"operation": "push", "repository": 5, "create_only": True},
+                "must list its branches or set branch_patterns",
+            ),
+            (
+                {"operation": "push", "repository": 5, "branches": ["a"], "branch_patterns": ["a"]},
+                "must not both list branches and set branch_patterns",
+            ),
+            ({"operation": "push", "repository": 5, "branch_patterns": []}, "must not be empty"),
+            ({"operation": "push", "repository": 5, "branch_patterns": [""]}, "must not be empty"),
+            ({"operation": "push", "repository": 5, "branch_patterns": ["seer/("]}, "is not a regular expression"),
+            ({"operation": "push", "repository": 5, "branches": ["a"], "create_only": 1}, "Expected `bool`"),
+            (
+                {"operation": "push", "repository": 5, "branches": ["a"], "any_object": True},
+                "unknown field `any_object`",
+            ),
+            ({"operation": "fetch", "repository": 5, "any_object": True, "create_only": True}, "unknown field"),
+            (
+                {"operation": "fetch", "repository": 5, "branch_patterns": ["seer/.+"]},
+                "unknown field `branch_patterns`",
+            ),
+            ({"operation": "fetch", "repository": 5, "branches": ["main"], "deepen": 50}, "unknown field `deepen`"),
+            ({"operation": "delete", "repository": 5, "branches": ["main"]}, "Invalid value 'delete'"),
+            ({"repository": 5, "any_object": True}, "missing required field `operation`"),
+            ({"operation": "fetch", "any_object": True}, "missing required field `repository`"),
+        ],
+    )
+    def test_invalid_rule(self, rule, error):
+        assert_invalid(sign_claims(rule_claims(rule)), error)
+
+    def test_rule_without_branches_or_any_object_cannot_be_built(self):
+        with pytest.raises(ValueError, match="must list its branches or set any_object"):
+            FetchRule(repository=5)
+
+    def test_rule_with_branches_and_any_object_cannot_be_built(self):
+        with pytest.raises(ValueError, match="must not both list branches and set any_object"):
+            FetchRule(repository=5, branches=["main"], any_object=True)
+
+    @pytest.mark.parametrize("make_rule", [FetchRule, PushRule], ids=["fetch", "push"])
+    def test_rule_without_branches_listed_cannot_be_built(self, make_rule):
+        with pytest.raises(ValueError, match="must not be empty"):
+            make_rule(repository=5, branches=[])
+
+    def test_no_rules(self):
+        with pytest.raises(ValueError, match="at least one rule"):
+            make_grant(rules=[])
+        assert_invalid(sign_claims(rule_claims()), "at least one rule")
+
+    def test_several_rules_for_one_repository_and_operation(self):
+        # A request passes if any of them allows it, so their order doesn't matter.
+        grant = make_grant(
+            rules=[
+                FetchRule(repository=5, any_object=True),
+                FetchRule(repository=5, branches=["main"]),
+                PushRule(repository=5, branch_patterns=["seer/.+"], create_only=True),
+                PushRule(repository=5, branches=["seer/fix"]),
+            ]
+        )
+        assert verify_grant([KEY.public], sign_grant(KEY.private, grant, now=NOW), now=NOW) == grant
+        assert grant.rules_for(5, "push") == grant.rules[2:]
+        assert grant.rules_for(("github", "123"), "push") == []
+
+    def test_one_repository_named_both_ways(self):
+        # Nothing matches a database id to a provider id, so the two are separate repositories here, as in requests.
+        rules = [FetchRule(repository=5, any_object=True), FetchRule(repository=("github", "123"), branches=["main"])]
+        assert make_grant(rules=rules).rules == rules
+
+
+INVALID_BRANCH_NAMES = [
+    "",
+    "refs/heads/main",
+    "refs/tags/v1",
+    "-main",
+    "HEAD",
+    "@",
+    "a..b",
+    "a b",
+    "a\tb",
+    "a\x00b",
+    "a\x7fb",
+    "a~1",
+    "a^",
+    "a:b",
+    "a?",
+    "a*",
+    "a[b",
+    "a\\b",
+    "/main",
+    "main/",
+    "a//b",
+    ".main",
+    "a/.b",
+    "main.lock",
+    "a/b.lock/c",
+    "main.",
+    "a@{1}",
+]
+
+
+class TestBranchPatterns:
+    @pytest.mark.parametrize(
+        "patterns,branch,covered",
+        [
+            (["seer/.+"], "seer/fix", True),
+            (["seer/.+"], "seer/fix/deeper", True),
+            (["seer/.+"], "seer/", False),
+            # Matched against the whole name, not searched for in it.
+            (["seer/.+"], "main-seer/fix", False),
+            (["seer/[a-z-]+"], "seer/fix/deeper", False),
+            (["seer"], "seer/fix", False),
+            (["seer/[a-z]+", "cursor/.+"], "cursor/x", True),
+            (["seer/.+"], "Seer/fix", False),
+        ],
+    )
+    def test_covers(self, patterns, branch, covered):
+        assert PushRule(repository=5, branch_patterns=patterns).covers(branch) is covered
+
+    def test_listed_branches(self):
+        rule = PushRule(repository=5, branches=["seer/fix"])
+        assert rule.covers("seer/fix")
+        assert not rule.covers("seer/fix2")
+
+    def test_cannot_be_built_with_an_invalid_pattern(self):
+        with pytest.raises(ValueError, match="'seer/\\(' is not a regular expression"):
+            PushRule(repository=5, branch_patterns=["seer/("])
+
+
+class TestBranchNames:
+    @pytest.mark.parametrize("name", INVALID_BRANCH_NAMES)
+    def test_invalid(self, name):
+        with pytest.raises(ValueError, match="is not a branch name"):
+            PushRule(repository=5, branches=["main", name])
+        assert_invalid(
+            sign_claims(rule_claims({"operation": "push", "repository": 5, "branches": ["main", name]})),
+            "is not a branch name",
+        )
+
+    def test_checked_in_fetch_rules_too(self):
+        with pytest.raises(ValueError, match="is not a branch name"):
+            FetchRule(repository=5, branches=["refs/heads/main"])
+
+    @pytest.mark.parametrize(
+        "name", ["main", "seer/fix-123", "feature/a.b", "a@b", "release-1.0", "a.lock.b", "x/HEAD", "ünicode"]
+    )
+    def test_valid(self, name):
+        assert PushRule(repository=5, branches=[name]).branches == [name]
+
+
+def sign_v1(grant: Grant) -> str:
+    return sign_grant(KEY.private, grant, now=NOW)
+
+
+class TestVersion1:
+    """A grant with operation and repositories instead of rules, as before rules, still signs and verifies."""
+
+    @pytest.mark.parametrize("operation", ["fetch", "push"])
+    def test_verifies_what_it_signed(self, operation):
+        grant = make_v1_grant(operation=operation)
+        verified = verify_grant([KEY.public], sign_v1(grant), now=NOW)
+        assert verified == grant
+        assert verified.version == 1
+        assert verified.rules is None
+
+    @pytest.mark.parametrize(
+        "fields,error",
+        [
+            ({"operation": "fetch"}, "must set rules, or both operation and repositories"),
+            ({"repositories": [5]}, "must set rules, or both operation and repositories"),
+            ({}, "must set rules, or both operation and repositories"),
+            (
+                {"operation": "fetch", "repositories": [5], "rules": [FetchRule(repository=5, any_object=True)]},
+                "either rules or operation and repositories, not both",
+            ),
+        ],
+    )
+    def test_one_form_or_the_other(self, fields, error):
+        with pytest.raises(ValueError, match=error):
+            Grant(organization_id=1, referrer="seer.coding_job", expires_at=FAR_FUTURE, **fields)
+
+    @pytest.mark.parametrize(
+        "change",
+        [
+            {"version": 2},
+            {"rules": [{"operation": "fetch", "repository": 5, "any_object": True}]},
+        ],
+    )
+    def test_version_must_match_the_form(self, change):
+        assert_invalid(sign_claims(claims_of(make_v1_grant()) | change), "payload is invalid")
+
+    def test_token_is_a_jwt_with_the_grants_claims(self):
+        token = sign_v1(make_v1_grant())
+        assert jwt.decode(token, options={"verify_signature": False}) == {
+            "organization_id": 1,
+            "referrer": "seer.coding_job",
+            "operation": "fetch",
+            "repositories": [5, ["github", "123"]],
+            "version": 1,
+            "exp": FAR_FUTURE,
+            "iat": NOW,
+            "aud": GRANT_AUDIENCE,
+        }
+
+    def test_without_a_version(self):
+        # Read as version 1, as before there was a second one.
+        claims = claims_of(make_v1_grant())
+        del claims["version"]
+        assert verify_grant([KEY.public], sign_claims(claims), now=NOW) == make_v1_grant()
+
+    def test_expired(self):
+        with pytest.raises(GitGrantExpired):
+            verify_grant([KEY.public], sign_v1(make_v1_grant(expires_at=NOW)), now=NOW)
+
+    @pytest.mark.parametrize(
+        "change",
+        [
+            {"operation": "delete"},
+            {"repositories": [["github"]]},
+            {"repositories": [["unknown_provider", "123"]]},
+            {"rules": [{"operation": "push", "repository": 5, "branches": ["main"]}]},
+        ],
+    )
+    def test_signed_but_invalid_claims(self, change):
+        assert_invalid(sign_claims(claims_of(make_v1_grant()) | change), "payload is invalid")
+
     def test_no_repositories(self):
-        assert_invalid(sign_grant(KEY.private, make_grant(repositories=[]), now=NOW), "names no repositories")
+        assert_invalid(sign_v1(make_v1_grant(repositories=[])), "names no repositories")
diff --git a/tests/unit/test_git_push.py b/tests/unit/test_git_push.py
new file mode 100644
index 00000000..5c461d22
--- /dev/null
+++ b/tests/unit/test_git_push.py
@@ -0,0 +1,355 @@
+import io
+
+import pytest
+
+from scm.errors import ClientClosedRequest, GitMalformedRequestBody, GitOperationNotGranted, RpcMalformedRequestHeaders
+from scm.git.push import (
+    MAX_UPDATES_LENGTH,
+    PushRequest,
+    RefUpdate,
+    ReplayedBody,
+    bounded_body,
+    read_push_request,
+    refusal_report,
+    refused_updates,
+)
+from scm.git.types import PushRule
+from tests.unit.git_builders import NEW_ID, OLD_ID, PACK, ZERO_ID, make_push_body, pkt
+
+RULES = [PushRule(repository=5, branches=["seer/fix", "seer/other"])]
+
+
+def read(body: bytes, headers: dict[str, str] | None = None) -> tuple[PushRequest, io.BytesIO]:
+    stream = io.BytesIO(body)
+    return read_push_request(headers or {}, stream), stream
+
+
+class Trickle(io.BytesIO):
+    """A body that returns at most one byte per read, as a socket may."""
+
+    def read(self, size: int | None = -1, /) -> bytes:
+        return super().read(1 if size is None or size < 0 else min(size, 1))
+
+
+class Failing:
+    def read(self, size: int = -1, /) -> bytes:
+        raise ConnectionResetError()
+
+
+class TestRead:
+    def test_updates_and_capabilities(self):
+        body = make_push_body((OLD_ID, NEW_ID, "refs/heads/seer/fix"), (ZERO_ID, NEW_ID, "refs/heads/seer/other"))
+        request, stream = read(body)
+        assert request.updates == [
+            RefUpdate(old_id=OLD_ID.encode(), new_id=NEW_ID.encode(), ref=b"refs/heads/seer/fix"),
+            RefUpdate(old_id=ZERO_ID.encode(), new_id=NEW_ID.encode(), ref=b"refs/heads/seer/other"),
+        ]
+        assert request.capabilities == {
+            "report-status",
+            "side-band-64k",
+            "quiet",
+            "object-format=sha1",
+            "agent=git/2.54.0",
+        }
+
+    def test_leaves_the_pack_unread(self):
+        body = make_push_body((OLD_ID, NEW_ID, "refs/heads/seer/fix"))
+        request, stream = read(body)
+        assert request.head + stream.read() == body
+        assert stream.getvalue()[len(request.head) :] == PACK
+
+    def test_probe(self):
+        # git sends a lone flush before a push too large to buffer, to check its credentials before it streams.
+        request, stream = read(b"0000")
+        assert request == PushRequest(updates=[], capabilities=frozenset(), head=b"0000")
+
+    def test_shallow_commits(self):
+        body = make_push_body((OLD_ID, NEW_ID, "refs/heads/seer/fix"), shallow=("3" * 40, "4" * 40))
+        request, stream = read(body)
+        assert [update.ref for update in request.updates] == [b"refs/heads/seer/fix"]
+        assert request.head + stream.read() == body
+
+    def test_sha256_object_ids(self):
+        old_id, new_id = "a" * 64, "b" * 64
+        request, _ = read(make_push_body((old_id, new_id, "refs/heads/seer/fix")))
+        assert request.updates[0].new_id == new_id.encode()
+
+    def test_short_reads(self):
+        body = make_push_body((OLD_ID, NEW_ID, "refs/heads/seer/fix"), shallow=("3" * 40,))
+        request = read_push_request({}, Trickle(body))
+        assert request.updates[0].ref == b"refs/heads/seer/fix"
+
+    def test_trailing_newlines(self):
+        line = f"{OLD_ID} {NEW_ID} refs/heads/seer/fix".encode()
+        request, _ = read(pkt(line + b"\0report-status\n") + pkt(line.replace(b"fix", b"other") + b"\n") + b"0000")
+        assert [update.ref for update in request.updates] == [b"refs/heads/seer/fix", b"refs/heads/seer/other"]
+        assert request.capabilities == {"report-status"}
+
+    def test_uppercase_pkt_length(self):
+        line = f"{OLD_ID} {NEW_ID} refs/heads/{'x' * 200}".encode()
+        request, _ = read(b"%04X" % (len(line) + 4) + line + b"0000")
+        assert request.updates[0].ref == f"refs/heads/{'x' * 200}".encode()
+
+    @pytest.mark.parametrize("encoding", ["identity", "", " Identity "])
+    def test_identity_encoding(self, encoding):
+        request, _ = read(make_push_body((OLD_ID, NEW_ID, "refs/heads/seer/fix")), {"Content-Encoding": encoding})
+        assert len(request.updates) == 1
+
+    @pytest.mark.parametrize("encoding", ["gzip", "deflate"])
+    def test_compressed(self, encoding):
+        with pytest.raises(GitOperationNotGranted, match="compressed push"):
+            read(make_push_body((OLD_ID, NEW_ID, "refs/heads/seer/fix")), {"content-encoding": encoding})
+
+    def test_signed_push(self):
+        body = pkt(b"push-cert\0report-status") + pkt(b"certificate version 0.1\n") + b"0000"
+        with pytest.raises(GitOperationNotGranted, match="signed pushes"):
+            read(body)
+
+    @pytest.mark.parametrize(
+        "body",
+        [
+            pytest.param(b"", id="empty"),
+            pytest.param(b"00", id="short-length"),
+            pytest.param(b"zzzz", id="length-not-hex"),
+            pytest.param(b"+004", id="length-with-sign"),
+            pytest.param(b" 0a\n", id="length-with-spaces"),
+            pytest.param(b"0001", id="delimiter"),
+            pytest.param(b"0003", id="v2-special-packet"),
+            pytest.param(b"fff1" + b"x" * 65517, id="pkt-line-too-long"),
+            pytest.param(pkt(f"{OLD_ID} {NEW_ID} refs/heads/a".encode()), id="no-flush"),
+            pytest.param(pkt(f"{OLD_ID} {NEW_ID}".encode()) + b"0000", id="no-ref"),
+            pytest.param(pkt(f"{OLD_ID} {NEW_ID} ".encode()) + b"0000", id="empty-ref"),
+            pytest.param(pkt(f"{OLD_ID[:-1]} {NEW_ID} refs/heads/a".encode()) + b"0000", id="short-id"),
+            pytest.param(pkt(f"{'A' * 40} {NEW_ID} refs/heads/a".encode()) + b"0000", id="uppercase-id"),
+            pytest.param(pkt(f"{OLD_ID} {'b' * 64} refs/heads/a".encode()) + b"0000", id="mixed-id-lengths"),
+            pytest.param(pkt(b"shallow 123\n") + b"0000", id="bad-shallow"),
+            pytest.param(pkt(f"{OLD_ID} {NEW_ID} refs/heads/a\0caps \xff".encode("latin-1")) + b"0000", id="caps"),
+        ],
+    )
+    def test_malformed(self, body):
+        with pytest.raises(GitMalformedRequestBody):
+            read(body)
+
+    def test_shallow_line_after_an_update(self):
+        body = make_push_body((OLD_ID, NEW_ID, "refs/heads/a"))
+        flush = body.index(b"0000")
+        with pytest.raises(GitMalformedRequestBody):
+            read(body[:flush] + pkt(f"shallow {'3' * 40}".encode()) + body[flush:])
+
+    def test_too_long_before_the_pack(self):
+        updates = [(OLD_ID, NEW_ID, f"refs/heads/branch-{i:05d}") for i in range(MAX_UPDATES_LENGTH // 90)]
+        with pytest.raises(GitMalformedRequestBody, match=f"longer than {MAX_UPDATES_LENGTH} bytes"):
+            read(make_push_body(*updates))
+
+    def test_connection_fails(self):
+        with pytest.raises(ClientClosedRequest):
+            read_push_request({}, Failing())
+
+
+def request_for(*updates: tuple[str, str, str], capabilities: str = "report-status side-band-64k") -> PushRequest:
+    return read(make_push_body(*updates, capabilities=capabilities))[0]
+
+
+class TestRefusedUpdates:
+    def test_listed_branches(self):
+        request = request_for((OLD_ID, NEW_ID, "refs/heads/seer/fix"), (ZERO_ID, NEW_ID, "refs/heads/seer/other"))
+        assert refused_updates(RULES, request) == {}
+
+    def test_force_push_to_a_listed_branch(self):
+        # Nothing here knows whether the new commit descends from the old one, so the provider's protection decides.
+        assert refused_updates(RULES, request_for(("3" * 40, NEW_ID, "refs/heads/seer/fix"))) == {}
+
+    def test_probe(self):
+        assert refused_updates(RULES, read(b"0000")[0]) == {}
+
+    @pytest.mark.parametrize(
+        "ref,reason",
+        [
+            ("refs/heads/main", "the git grant does not cover this branch"),
+            ("refs/heads/seer/fix/deeper", "the git grant does not cover this branch"),
+            ("refs/heads/seer", "the git grant does not cover this branch"),
+            ("refs/heads/Seer/fix", "the git grant does not cover this branch"),
+            ("refs/tags/seer/fix", "the git grant allows pushing branches only"),
+            ("refs/pull/1/head", "the git grant allows pushing branches only"),
+            ("refs/notes/commits", "the git grant allows pushing branches only"),
+            ("HEAD", "the git grant allows pushing branches only"),
+            ("seer/fix", "the git grant allows pushing branches only"),
+        ],
+    )
+    def test_refs_the_rule_does_not_list(self, ref, reason):
+        assert refused_updates(RULES, request_for((OLD_ID, NEW_ID, ref))) == {ref.encode(): reason}
+
+    def test_deleting_a_listed_branch(self):
+        refused = refused_updates(RULES, request_for((OLD_ID, ZERO_ID, "refs/heads/seer/fix")))
+        assert refused == {b"refs/heads/seer/fix": "the git grant does not allow deleting a branch"}
+
+    def test_one_refused_update_refuses_the_push(self):
+        request = request_for((OLD_ID, NEW_ID, "refs/heads/seer/fix"), (OLD_ID, NEW_ID, "refs/heads/main"))
+        assert refused_updates(RULES, request) == {
+            b"refs/heads/main": "the git grant does not cover this branch",
+            b"refs/heads/seer/fix": "refused with the rest of the push",
+        }
+
+    def test_no_rules(self):
+        refused = refused_updates([], request_for((OLD_ID, NEW_ID, "refs/heads/seer/fix")))
+        assert refused == {b"refs/heads/seer/fix": "the git grant does not cover this branch"}
+
+    @pytest.mark.parametrize(
+        "old_id,ref,reason",
+        [
+            (ZERO_ID, "refs/heads/seer/new", None),
+            (OLD_ID, "refs/heads/seer/new", "the git grant allows only creating this branch"),
+            (ZERO_ID, "refs/heads/main", "the git grant does not cover this branch"),
+            (ZERO_ID, "refs/tags/seer/new", "the git grant allows pushing branches only"),
+        ],
+    )
+    def test_create_only_pattern(self, old_id, ref, reason):
+        rules = [PushRule(repository=5, branch_patterns=["seer/.+"], create_only=True)]
+        refused = refused_updates(rules, request_for((old_id, NEW_ID, ref)))
+        assert refused == ({} if reason is None else {ref.encode(): reason})
+
+    def test_create_only_listed_branch(self):
+        rules = [PushRule(repository=5, branches=["seer/fix"], create_only=True)]
+        assert refused_updates(rules, request_for((ZERO_ID, NEW_ID, "refs/heads/seer/fix"))) == {}
+        assert refused_updates(rules, request_for((OLD_ID, NEW_ID, "refs/heads/seer/fix"))) == {
+            b"refs/heads/seer/fix": "the git grant allows only creating this branch"
+        }
+
+    def test_non_utf8_branch_against_a_pattern(self):
+        rules = [PushRule(repository=5, branch_patterns=[".+"])]
+        ref = b"refs/heads/\xff"
+        request = PushRequest(
+            updates=[RefUpdate(old_id=OLD_ID.encode(), new_id=NEW_ID.encode(), ref=ref)],
+            capabilities=frozenset(),
+            head=b"",
+        )
+        assert refused_updates(rules, request) == {ref: "the git grant does not cover this branch"}
+
+    def test_any_rule_allows_an_update(self):
+        # The agent may start branches under seer/ and keep pushing to the one Seer added after its first push.
+        rules = [
+            PushRule(repository=5, branch_patterns=["seer/.+"], create_only=True),
+            PushRule(repository=5, branches=["seer/fix"]),
+        ]
+        request = request_for((OLD_ID, NEW_ID, "refs/heads/seer/fix"), (ZERO_ID, NEW_ID, "refs/heads/seer/next"))
+        assert refused_updates(rules, request) == {}
+        assert refused_updates(list(reversed(rules)), request) == {}
+
+    def test_reason_of_the_rule_the_update_came_closest_to_passing(self):
+        rules = [
+            PushRule(repository=5, branches=["other"]),
+            PushRule(repository=5, branch_patterns=["seer/.+"], create_only=True),
+        ]
+        assert refused_updates(rules, request_for((OLD_ID, NEW_ID, "refs/heads/seer/old"))) == {
+            b"refs/heads/seer/old": "the git grant allows only creating this branch"
+        }
+        assert refused_updates(rules, request_for((OLD_ID, ZERO_ID, "refs/heads/seer/old"))) == {
+            b"refs/heads/seer/old": "the git grant does not allow deleting a branch"
+        }
+
+    def test_push_options(self):
+        request = request_for((OLD_ID, NEW_ID, "refs/heads/seer/fix"), capabilities="report-status push-options")
+        assert refused_updates(RULES, request) == {
+            b"refs/heads/seer/fix": "the git endpoint does not relay push options"
+        }
+
+
+def unband(response: bytes) -> bytes:
+    """The band 1 data of a side band response, checking that it ends with a flush."""
+    data, position = b"", 0
+    while response[position : position + 4] != b"0000":
+        length = int(response[position : position + 4], 16)
+        assert response[position + 4] == 1
+        data += response[position + 5 : position + length]
+        position += length
+    assert position + 4 == len(response)
+    return data
+
+
+class TestRefusalReport:
+    REFUSED = {b"refs/heads/main": "the git grant does not cover this branch"}
+    REPORT = pkt(b"unpack ok\n") + pkt(b"ng refs/heads/main the git grant does not cover this branch\n") + b"0000"
+
+    def test_report_status(self):
+        request = request_for((OLD_ID, NEW_ID, "refs/heads/main"), capabilities="report-status")
+        assert refusal_report(request, self.REFUSED) == self.REPORT
+
+    def test_report_status_v2(self):
+        request = request_for((OLD_ID, NEW_ID, "refs/heads/main"), capabilities="report-status-v2")
+        assert refusal_report(request, self.REFUSED) == self.REPORT
+
+    def test_side_band_64k(self):
+        request = request_for((OLD_ID, NEW_ID, "refs/heads/main"), capabilities="report-status side-band-64k")
+        response = refusal_report(request, self.REFUSED)
+        assert response == pkt(b"\1" + self.REPORT) + b"0000"
+
+    def test_side_band_splits_at_1000_bytes(self):
+        updates = [(OLD_ID, NEW_ID, f"refs/heads/branch-{i:03d}") for i in range(40)]
+        request = request_for(*updates, capabilities="report-status side-band")
+        refused = {update.ref: "the git grant does not cover this branch" for update in request.updates}
+        response = refusal_report(request, refused)
+        assert response is not None
+        assert max(int(response[i : i + 4], 16) for i in range(0, len(response) - 4, 1000)) <= 1000
+        assert unband(response).startswith(pkt(b"unpack ok\n"))
+        assert unband(response).count(b"ng refs/heads/branch-") == 40
+
+    def test_every_update_in_order(self):
+        request = request_for((OLD_ID, NEW_ID, "refs/heads/seer/fix"), (OLD_ID, NEW_ID, "refs/heads/main"))
+        response = refusal_report(request, refused_updates(RULES, request))
+        assert response is not None
+        assert unband(response) == (
+            pkt(b"unpack ok\n")
+            + pkt(b"ng refs/heads/seer/fix refused with the rest of the push\n")
+            + pkt(b"ng refs/heads/main the git grant does not cover this branch\n")
+            + b"0000"
+        )
+
+    def test_client_asked_for_no_report(self):
+        request = request_for((OLD_ID, NEW_ID, "refs/heads/main"), capabilities="side-band-64k quiet")
+        assert refusal_report(request, self.REFUSED) is None
+
+
+class TestReplayedBody:
+    @pytest.mark.parametrize("size", [1, 3, 5, 64, 1024])
+    def test_head_then_rest(self, size):
+        body = ReplayedBody(b"head", io.BytesIO(b"rest of the body"))
+        chunks = []
+        while chunk := body.read(size):
+            chunks.append(chunk)
+        assert b"".join(chunks) == b"headrest of the body"
+        assert all(len(chunk) <= size for chunk in chunks)
+
+    def test_read_all(self):
+        assert ReplayedBody(b"head", io.BytesIO(b"rest")).read() == b"headrest"
+
+    def test_empty_head(self):
+        assert ReplayedBody(b"", io.BytesIO(b"rest")).read(64) == b"rest"
+
+
+class TestBoundedBody:
+    def test_ends_at_content_length(self):
+        # As a connection's socket would, the stream goes on past the body.
+        stream = io.BytesIO(b"body" + b"next request")
+        body = bounded_body({"content-length": "4"}, stream)
+        assert body.read(64) == b"body"
+        assert body.read(64) == b""
+        assert body.read() == b""
+        assert stream.tell() == 4
+
+    def test_short_reads(self):
+        body = bounded_body({"Content-Length": "4"}, Trickle(b"body and more"))
+        assert b"".join(iter(lambda: body.read(64), b"")) == b"body"
+
+    def test_without_content_length(self):
+        stream = io.BytesIO(b"body")
+        assert bounded_body({}, stream) is stream
+
+    @pytest.mark.parametrize("length", ["x", "-1", ""])
+    def test_invalid_content_length(self, length):
+        with pytest.raises(RpcMalformedRequestHeaders):
+            bounded_body({"Content-Length": length}, io.BytesIO(b""))
+
+    def test_truncated_updates(self):
+        body = make_push_body((OLD_ID, NEW_ID, "refs/heads/seer/fix"))
+        with pytest.raises(GitMalformedRequestBody, match="ended before its ref updates did"):
+            read_push_request({}, bounded_body({"Content-Length": "20"}, io.BytesIO(body)))
diff --git a/tests/unit/test_git_relay.py b/tests/unit/test_git_relay.py
index fe304ed5..99833a36 100644
--- a/tests/unit/test_git_relay.py
+++ b/tests/unit/test_git_relay.py
@@ -26,10 +26,11 @@
     SCMCodedError,
 )
 from scm.git.relay import GitRelay, _UpstreamStream, authorize_request
+from scm.git.types import FetchRule, PushRule
 from scm.secret import Secret
 from scm.types import GitRemote
 from tests.unit.error_reports import assert_not_in_frames, assert_not_reported, reported_local
-from tests.unit.git_builders import make_grant
+from tests.unit.git_builders import make_grant, make_v1_grant
 from tests.unit.git_upstream import (
     ClosingAdapter,
     HostRefused,
@@ -741,24 +742,57 @@ def handler(h: UpstreamHandler) -> None:
 
 FETCH_ROUTES = [("GET", "info/refs", "service=git-upload-pack"), ("POST", "git-upload-pack", "")]
 PUSH_ROUTES = [("GET", "info/refs", "service=git-receive-pack"), ("POST", "git-receive-pack", "")]
+FETCH_ANY = FetchRule(repository=5, any_object=True)
+PUSH_BRANCH = PushRule(repository=5, branches=["seer/fix"])
 
 
 class TestAuthorizeRequest:
-    @pytest.mark.parametrize(
-        "operation,route",
-        [("fetch", route) for route in FETCH_ROUTES] + [("push", route) for route in PUSH_ROUTES],
-    )
-    def test_grant_reaches_its_operations_routes(self, operation, route):
-        assert authorize_request(make_grant(operation=operation), 5, *route) == operation
+    @pytest.mark.parametrize("route", FETCH_ROUTES)
+    def test_fetch_rule_reaches_fetch_routes(self, route):
+        assert authorize_request(make_grant(rules=[FETCH_ANY]), 5, *route) == "fetch"
+
+    @pytest.mark.parametrize("route", PUSH_ROUTES)
+    def test_push_rule_reaches_push_routes(self, route):
+        # GitServer keeps the push to the rule's branches.
+        assert authorize_request(make_grant(rules=[PUSH_BRANCH]), 5, *route) == "push"
+
+    @pytest.mark.parametrize("route", PUSH_ROUTES)
+    def test_fetch_rule_cannot_reach_push_routes(self, route):
+        with pytest.raises(GitOperationNotGranted, match="allows fetch on this repository, not push"):
+            authorize_request(make_grant(rules=[FETCH_ANY]), 5, *route)
+
+    @pytest.mark.parametrize("route", FETCH_ROUTES)
+    def test_push_rule_cannot_reach_fetch_routes(self, route):
+        with pytest.raises(GitOperationNotGranted, match="allows push on this repository, not fetch"):
+            authorize_request(make_grant(rules=[PUSH_BRANCH]), 5, *route)
+
+    def test_rule_for_another_repository_does_not_apply(self):
+        grant = make_grant(rules=[FETCH_ANY, PushRule(repository=6, branches=["seer/fix"])])
+        with pytest.raises(GitOperationNotGranted, match="allows fetch on this repository, not push"):
+            authorize_request(grant, 5, "POST", "git-receive-pack", "")
+
+    def test_any_of_several_rules(self):
+        grant = make_grant(
+            rules=[
+                FetchRule(repository=5, branches=["main"]),
+                FETCH_ANY,
+                PUSH_BRANCH,
+                PushRule(repository=5, branch_patterns=["seer/.+"], create_only=True),
+            ]
+        )
+        assert authorize_request(grant, 5, "POST", "git-upload-pack", "") == "fetch"
+        assert authorize_request(grant, 5, "POST", "git-receive-pack", "") == "push"
 
-    @pytest.mark.parametrize(
-        "operation,route",
-        [("fetch", route) for route in PUSH_ROUTES] + [("push", route) for route in FETCH_ROUTES],
-    )
-    def test_grant_cannot_reach_the_other_operations_routes(self, operation, route):
-        other = "push" if operation == "fetch" else "fetch"
-        with pytest.raises(GitOperationNotGranted, match=f"allows {operation}, not {other}"):
-            authorize_request(make_grant(operation=operation), 5, *route)
+    def test_fetch_rules_that_all_list_branches_are_refused(self):
+        grant = make_grant(rules=[FetchRule(repository=5, branches=["main"]), FetchRule(repository=5, branches=["a"])])
+        with pytest.raises(GitOperationNotGranted, match="cannot keep a fetch to a rule's branches"):
+            authorize_request(grant, 5, "POST", "git-upload-pack", "")
+
+    @pytest.mark.parametrize("route", FETCH_ROUTES)
+    def test_fetch_rule_that_lists_branches_is_refused(self, route):
+        # Nothing reads a fetch's body, so nothing would keep the fetch to the rule's branches.
+        with pytest.raises(GitOperationNotGranted, match="cannot keep a fetch to a rule's branches"):
+            authorize_request(make_grant(rules=[FetchRule(repository=5, branches=["main"])]), 5, *route)
 
     @pytest.mark.parametrize(
         "method,route,query",
@@ -779,23 +813,57 @@ def test_grant_cannot_reach_the_other_operations_routes(self, operation, route):
         ],
     )
     def test_routes_that_are_not_git(self, method, route, query):
-        for operation in ("fetch", "push"):
-            with pytest.raises(GitRouteNotSupported, match="not a git fetch or push route"):
-                authorize_request(make_grant(operation=operation), 5, method, route, query)
+        with pytest.raises(GitRouteNotSupported, match="not a git fetch or push route"):
+            authorize_request(make_grant(rules=[FETCH_ANY, PUSH_BRANCH]), 5, method, route, query)
 
     def test_repository_not_in_grant(self):
         with pytest.raises(GitRepositoryNotGranted, match="does not cover this repository"):
             authorize_request(make_grant(), 6, "POST", "git-upload-pack", "")
 
     def test_repository_named_another_way_is_not_in_grant(self):
-        # The grant and the request name repositories the same way; matching a database id to a provider id
-        # would mean resolving every repository the grant lists.
-        grant = make_grant(repositories=[("github", "123")])
+        # The rule and the request must name a repository the same way, since matching a database id to a provider
+        # id would mean resolving every repository the grant names.
+        grant = make_grant(rules=[FetchRule(repository=("github", "123"), any_object=True)])
         with pytest.raises(GitRepositoryNotGranted):
             authorize_request(grant, 5, "POST", "git-upload-pack", "")
         assert authorize_request(grant, ("github", "123"), "POST", "git-upload-pack", "") == "fetch"
 
+    def test_any_repository_with_a_rule(self):
+        grant = make_grant(
+            rules=[
+                FetchRule(repository=repository, any_object=True) for repository in (5, 6, ("gitlab", "gitlab.com:7"))
+            ]
+        )
+        assert authorize_request(grant, 6, "POST", "git-upload-pack", "") == "fetch"
+        assert authorize_request(grant, ("gitlab", "gitlab.com:7"), "POST", "git-upload-pack", "") == "fetch"
+
+
+class TestAuthorizeRequestVersion1:
+    @pytest.mark.parametrize(
+        "operation,route",
+        [("fetch", route) for route in FETCH_ROUTES] + [("push", route) for route in PUSH_ROUTES],
+    )
+    def test_grant_reaches_its_operations_routes(self, operation, route):
+        assert authorize_request(make_v1_grant(operation=operation), 5, *route) == operation
+
+    @pytest.mark.parametrize(
+        "operation,route",
+        [("fetch", route) for route in PUSH_ROUTES] + [("push", route) for route in FETCH_ROUTES],
+    )
+    def test_grant_cannot_reach_the_other_operations_routes(self, operation, route):
+        other = "push" if operation == "fetch" else "fetch"
+        with pytest.raises(GitOperationNotGranted, match=f"allows {operation}, not {other}"):
+            authorize_request(make_v1_grant(operation=operation), 5, *route)
+
+    def test_route_that_is_not_git(self):
+        with pytest.raises(GitRouteNotSupported):
+            authorize_request(make_v1_grant(), 5, "POST", "git-upload-archive", "")
+
+    def test_repository_not_in_grant(self):
+        with pytest.raises(GitRepositoryNotGranted, match="does not cover this repository"):
+            authorize_request(make_v1_grant(), 6, "POST", "git-upload-pack", "")
+
     def test_any_listed_repository(self):
-        grant = make_grant(repositories=[5, 6, ("gitlab", "gitlab.com:7")])
+        grant = make_v1_grant(repositories=[5, 6, ("gitlab", "gitlab.com:7")])
         assert authorize_request(grant, 6, "POST", "git-upload-pack", "") == "fetch"
         assert authorize_request(grant, ("gitlab", "gitlab.com:7"), "POST", "git-upload-pack", "") == "fetch"
diff --git a/tests/unit/test_git_server.py b/tests/unit/test_git_server.py
index b6dc082e..98f4a0c8 100644
--- a/tests/unit/test_git_server.py
+++ b/tests/unit/test_git_server.py
@@ -20,12 +20,22 @@
 from scm.git.grant import sign_grant
 from scm.git.relay import GitRelay, RelayResponse, ResponseBody
 from scm.git.server import _STATUS, GitServer
-from scm.git.types import GRANT_HEADER, GitOperation, Readable
+from scm.git.types import GRANT_HEADER, FetchRule, GitOperation, PushRule, Readable
 from scm.rpc.types import StreamResponse
 from scm.secret import Secret
 from scm.types import GitRemote, Provider, Repository, RepositoryId
 from tests.unit.error_reports import assert_not_in_frames, assert_not_reported, reported_local
-from tests.unit.git_builders import make_grant, make_key_pair, make_provider, make_repository
+from tests.unit.git_builders import (
+    NEW_ID,
+    OLD_ID,
+    make_grant,
+    make_key_pair,
+    make_provider,
+    make_push_body,
+    make_repository,
+    make_v1_grant,
+    pkt,
+)
 from tests.unit.git_upstream import (
     ClosingAdapter,
     HostRefused,
@@ -226,7 +236,7 @@ def test_route_that_is_not_git(self):
 
     def test_operation_the_grant_does_not_allow(self):
         recorder = Recorder()
-        assert_refused(push(make_server(recorder), signed_grant()), 403, "allows fetch, not push")
+        assert_refused(push(make_server(recorder), signed_grant()), 403, "allows fetch on this repository, not push")
         assert recorder.repository_lookups == []
         assert recorder.token_lookups == []
         assert recorder.counts == [
@@ -234,6 +244,13 @@ def test_operation_the_grant_does_not_allow(self):
         ]
         assert recorder.errors[0][1] == {"repository_id": "5", "organization_id": "1", "referrer": REFERRER}
 
+    def test_fetch_rule_that_lists_branches(self):
+        recorder = Recorder()
+        grant = signed_grant(rules=[FetchRule(repository=5, branches=["main"])])
+        assert_refused(request(make_server(recorder), grant), 403, "cannot keep a fetch to a rule's branches")
+        assert recorder.repository_lookups == []
+        assert recorder.token_lookups == []
+
     def test_repository_the_grant_does_not_list(self):
         recorder = Recorder()
         assert_refused(request(make_server(recorder), signed_grant(), repository_id=6), 403, "does not cover")
@@ -296,10 +313,10 @@ def test_no_token(self):
             ("sentry.scm.git.request.failed", {"referrer": REFERRER, "code": "git_token_unavailable"})
         ]
 
-    def test_token_asked_for_the_grants_operation(self):
+    def test_token_asked_for_the_requests_operation(self):
         recorder = Recorder()
-        push(make_server(recorder), signed_grant(operation="push"))
-        assert recorder.token_lookups == [(make_repository(), "push")]
+        request(make_server(recorder), signed_grant())
+        assert recorder.token_lookups == [(make_repository(), "fetch")]
 
     def test_forwards_with_remote_and_token(self):
         recorder = Recorder()
@@ -618,17 +635,17 @@ def test_counts_bytes_each_way_end_to_end(self):
             relay = GitRelay(ca_bundle=upstream.ca_bundle)
             response = request(
                 make_server(recorder, relay),
-                signed_grant(operation="push"),
+                signed_grant(),
                 method="POST",
-                route="git-receive-pack",
+                route="git-upload-pack",
                 query="",
                 headers={"Content-Length": "8"},
                 body=io.BytesIO(b"00000000"),
             )
             assert b"".join(response.content) == b"00000000"  # The fake upstream echoes the body.
         assert recorder.bytes == {
-            ("sentry.scm.git.request_size", "push"): 8,
-            ("sentry.scm.git.response_size", "push"): 8,
+            ("sentry.scm.git.request_size", "fetch"): 8,
+            ("sentry.scm.git.response_size", "fetch"): 8,
         }
 
     def test_relays_end_to_end(self):
@@ -721,3 +738,160 @@ def test_token_that_is_not_a_secret_is_refused_and_left_out(self):
         assert recorder.counts == [
             ("sentry.scm.git.request.failed", {"referrer": REFERRER, "code": "unhandled_exception"})
         ]
+
+
+PUSH_GRANT = sign_grant(KEY.private, make_grant(rules=[PushRule(repository=5, branches=["seer/fix"])]))
+
+
+def push_body(ref: str, capabilities: str = "report-status side-band-64k") -> bytes:
+    return make_push_body((OLD_ID, NEW_ID, ref), capabilities=capabilities)
+
+
+class Unreliable(io.BytesIO):
+    def read(self, size: int | None = -1, /) -> bytes:
+        raise ConnectionResetError()
+
+
+class TestPushRules:
+    """A push under a push rule reaches the provider only if the rule allows every ref it updates."""
+
+    def send(self, server: GitServer, body: Readable, headers: dict[str, str] | None = None) -> StreamResponse:
+        return request(
+            server, PUSH_GRANT, method="POST", route="git-receive-pack", query="", headers=headers, body=body
+        )
+
+    @pytest.mark.parametrize("sized", [True, False], ids=["content-length", "chunked"])
+    def test_listed_branch_end_to_end(self, sized):
+        body = push_body("refs/heads/seer/fix")
+        with running_upstream() as upstream:
+            recorder = Recorder(
+                repository=make_repository(provider_name="github_enterprise", web_base_url=upstream.url)
+            )
+            response = self.send(
+                make_server(recorder, GitRelay(ca_bundle=upstream.ca_bundle)),
+                io.BytesIO(body),
+                headers={"Content-Length": str(len(body))} if sized else None,
+            )
+            # The fake upstream echoes the body, so the provider got the updates the server read and the pack after.
+            assert b"".join(response.content) == body
+        assert recorder.token_lookups == [(recorder.repository, "push")]
+        assert recorder.bytes[("sentry.scm.git.request_size", "push")] == len(body)
+
+    def test_probe(self):
+        recorder = Recorder()
+        server = make_server(recorder)
+        assert self.send(server, io.BytesIO(b"0000")).status_code == 200
+        assert server.relay.forward.call_args.kwargs["body"].read() == b"0000"  # type: ignore[attr-defined]
+
+    def test_info_refs(self):
+        recorder = Recorder()
+        response = request(make_server(recorder), PUSH_GRANT, query="service=git-receive-pack")
+        assert b"".join(response.content) == b"refs"
+
+    def test_unlisted_branch_gets_a_report(self):
+        recorder = Recorder()
+        server = make_server(recorder)
+        body = io.BytesIO(push_body("refs/heads/main"))
+        response = self.send(server, body)
+        assert response.status_code == 200
+        assert response.headers["Content-Type"] == "application/x-git-receive-pack-result"
+        report = b"".join(response.content)
+        assert int(response.headers["Content-Length"]) == len(report)
+        assert report == (
+            pkt(
+                b"\1"
+                + pkt(b"unpack ok\n")
+                + pkt(b"ng refs/heads/main the git grant does not cover this branch\n")
+                + b"0000"
+            )
+            + b"0000"
+        )
+        # The pack is read to the end, so git gets to read the report rather than a broken pipe.
+        assert body.read() == b""
+        assert recorder.repository_lookups == []
+        assert recorder.token_lookups == []
+        server.relay.forward.assert_not_called()  # type: ignore[attr-defined]
+        assert recorder.counts == [
+            ("sentry.scm.git.request.failed", {"referrer": REFERRER, "code": "git_operation_not_granted"})
+        ]
+        [(error, context)] = recorder.errors
+        assert error.detail == (  # type: ignore[attr-defined]
+            "The git grant does not allow this push: refs/heads/main (the git grant does not cover this branch)."
+        )
+        assert context == {"repository_id": "5", "organization_id": "1", "referrer": REFERRER, "operation": "push"}
+
+    @pytest.mark.parametrize(
+        "ref,old_id,new_id,reason",
+        [
+            ("refs/tags/v1", OLD_ID, NEW_ID, "allows pushing branches only"),
+            ("refs/heads/seer/fix", OLD_ID, "0" * 40, "does not allow deleting a branch"),
+        ],
+    )
+    def test_other_refused_updates(self, ref, old_id, new_id, reason):
+        body = make_push_body((old_id, new_id, ref), capabilities="report-status")
+        assert reason.encode() in b"".join(self.send(make_server(Recorder()), io.BytesIO(body)).content)
+
+    def test_drains_no_further_than_content_length(self):
+        # A caller may pass a connection's stream, which goes on past the body, and reading on would wait for ever.
+        body = push_body("refs/heads/main")
+        stream = io.BytesIO(body + b"the next request")
+        response = self.send(make_server(Recorder()), stream, headers={"Content-Length": str(len(body))})
+        assert response.status_code == 200
+        assert stream.tell() == len(body)
+
+    def test_create_only_pattern_and_listed_branch(self):
+        grant = sign_grant(
+            KEY.private,
+            make_grant(
+                rules=[
+                    PushRule(repository=5, branch_patterns=["seer/.+"], create_only=True),
+                    PushRule(repository=5, branches=["seer/fix"]),
+                ]
+            ),
+        )
+        server = make_server(Recorder())
+        for body in (
+            make_push_body(("0" * 40, NEW_ID, "refs/heads/seer/next")),
+            make_push_body((OLD_ID, NEW_ID, "refs/heads/seer/fix")),
+        ):
+            response = request(server, grant, method="POST", route="git-receive-pack", query="", body=io.BytesIO(body))
+            assert response.status_code == 200
+        assert server.relay.forward.call_count == 2  # type: ignore[attr-defined]
+        refused = make_push_body((OLD_ID, NEW_ID, "refs/heads/seer/next"), capabilities="report-status")
+        response = request(server, grant, method="POST", route="git-receive-pack", query="", body=io.BytesIO(refused))
+        assert b"the git grant allows only creating this branch" in b"".join(response.content)
+
+    def test_client_that_asked_for_no_report(self):
+        recorder = Recorder()
+        body = io.BytesIO(push_body("refs/heads/main", capabilities="side-band-64k"))
+        assert_refused(self.send(make_server(recorder), body), 403, "does not allow this push: refs/heads/main")
+        assert body.read() == b""
+
+    def test_malformed(self):
+        recorder = Recorder()
+        assert_refused(self.send(make_server(recorder), io.BytesIO(b"garbage")), 400, "not pkt-lines")
+        assert recorder.repository_lookups == []
+        assert recorder.counts == [
+            ("sentry.scm.git.request.failed", {"referrer": REFERRER, "code": "git_malformed_request_body"})
+        ]
+
+    def test_compressed(self):
+        response = self.send(
+            make_server(Recorder()), io.BytesIO(push_body("refs/heads/seer/fix")), headers={"Content-Encoding": "gzip"}
+        )
+        assert_refused(response, 403, "compressed push")
+
+    def test_client_closes_while_sending_its_updates(self):
+        recorder = Recorder()
+        response = self.send(make_server(recorder), Unreliable())
+        assert response.status_code == 499
+        assert recorder.errors == []
+
+    def test_version_1_push_grant_relays_unread(self):
+        recorder = Recorder()
+        server = make_server(recorder)
+        grant = sign_grant(KEY.private, make_v1_grant(operation="push"))
+        body = io.BytesIO(b"not a push the server could read")
+        request(server, grant, method="POST", route="git-receive-pack", query="", body=body)
+        assert body.tell() == 0
+        assert server.relay.forward.call_args.kwargs["body"].read() == b"not a push the server could read"  # type: ignore[attr-defined]