defmodule OpenAgents.Forge.GitHTTPTest do
@moduledoc """
End-to-end forge tests with the real git client over real HTTP: clone,
push, WAL persistence, receipt derivation, re-materialization from the
WAL after cache loss, and auth refusals. This is the P1 exit test from
the roadmap, minus the fleet.
"""
use OpenAgents.DataCase, async: false
import OpenAgents.AccountsFixtures
alias OpenAgents.{AuditEvent, Forge, Machines, Repo, Repositories}
alias OpenAgents.Forge.{CacheReadiness, Repos, WAL}
defmodule TestPipeline do
@moduledoc false
use Plug.Builder
plug OpenAgentsWeb.Plugs.ForgeGitAuth
plug OpenAgents.Forge.GitHTTP
end
setup do
Ecto.Adapters.SQL.Sandbox.mode(OpenAgents.Repo, {:shared, self()})
base = Path.join(System.tmp_dir!(), "forge-e2e-#{System.unique_integer([:positive])}")
File.mkdir_p!(base)
previous_data = Application.get_env(:openagents, :forge_data_dir)
previous_wal = Application.get_env(:openagents, :forge_wal_dir)
Application.put_env(:openagents, :forge_data_dir, Path.join(base, "data"))
Application.put_env(:openagents, :forge_wal_dir, Path.join(base, "wal"))
CacheReadiness.reset()
user = repository_user_fixture("git-http-owner")
{:ok, repository, :created} =
OpenAgents.Repositories.create_user_repository(user, %{name: "demo"}, "git-http-demo")
repository =
repository
|> Ecto.Changeset.change(lifecycle_state: "ready", ready_at: DateTime.utc_now())
|> OpenAgents.Repo.update!()
{:ok, api_token, plaintext} =
OpenAgents.ApiTokens.create(user, %{
name: "Git HTTP test",
scopes: ["forge:write"],
lifetime_days: 1
})
port = free_port()
start_supervised!({Bandit, plug: TestPipeline, port: port, ip: {127, 0, 0, 1}})
on_exit(fn ->
Application.put_env(:openagents, :forge_data_dir, previous_data)
Application.put_env(:openagents, :forge_wal_dir, previous_wal)
CacheReadiness.reset()
File.rm_rf(base)
end)
%{
base: base,
port: port,
repository: repository,
api_token: api_token,
token: plaintext,
url: "http://x:#{plaintext}@127.0.0.1:#{port}/git-http-owner/demo.git",
user: user
}
end
defp free_port do
{:ok, socket} = :gen_tcp.listen(0, [])
{:ok, port} = :inet.port(socket)
:gen_tcp.close(socket)
port
end
# Every git call disables credential helpers so the developer keychain can
# neither pollute these tests nor be polluted by them.
defp sh!(dir, "git", args), do: sh_raw!(dir, "git", ["-c", "credential.helper="] ++ args)
defp sh!(dir, command, args), do: sh_raw!(dir, command, args)
defp sh_raw!(dir, command, args) do
{output, status} = System.cmd(command, args, cd: dir, stderr_to_stdout: true)
if status != 0, do: flunk("#{command} #{Enum.join(args, " ")} failed:\n#{output}")
output
end
defp seed_clone!(base, url) do
work = Path.join(base, "clone-#{System.unique_integer([:positive])}")
sh!(base, "git", ["clone", url, work])
sh!(work, "git", ["config", "user.email", "test@example.com"])
sh!(work, "git", ["config", "user.name", "Forge Test"])
work
end
defp commit_and_push!(work, filename, contents, message) do
File.write!(Path.join(work, filename), contents)
sh!(work, "git", ["add", "."])
sh!(work, "git", ["commit", "-m", message])
sh!(work, "git", ["push", "origin", "HEAD:main"])
end
test "clone, push, WAL persist, receipt, and PubSub — the ack chain", %{
base: base,
repository: repository,
url: url,
user: user
} do
Phoenix.PubSub.subscribe(OpenAgents.PubSub, "forge:pushes")
Repositories.subscribe_repository_changes()
stale_at = DateTime.add(DateTime.utc_now(), -86_400, :second)
Repo.update_all(
from(candidate in OpenAgents.Repositories.Repository,
where: candidate.id == ^repository.id
),
set: [updated_at: stale_at]
)
work = seed_clone!(base, url)
commit_and_push!(work, "hello.txt", "hello forge\n", "first commit")
# WAL is the authority: index has the entry and the ref.
assert {:ok, _generation, index} = WAL.read_index(repository.storage_key)
assert [entry] = WAL.entries(index)
assert entry["seq"] == 0
assert entry["principal"] == "user:#{user.id}"
assert %{"refs/heads/main" => sha} = WAL.refs(index)
assert byte_size(sha) == 40
# Derived receipt exists, idempotent by (repo, wal_seq).
assert [receipt] = Forge.recent_pushes(repository.storage_key)
assert receipt.wal_seq == 0
assert receipt.refs["refs/heads/main"]["new"] == sha
assert receipt.refs["refs/heads/main"]["old"] == nil
assert [logical_receipt] = Forge.recent_pushes(repository.name)
assert logical_receipt.id == receipt.id
# Deploy signal fired.
assert_receive {:forge_push, %{repo: storage_key, wal_seq: 0}}, 2_000
assert storage_key == repository.storage_key
assert_receive {:repository_changed, repository_id}, 2_000
assert repository_id == repository.id
refreshed_repository = Repo.get!(OpenAgents.Repositories.Repository, repository.id)
assert DateTime.after?(refreshed_repository.updated_at, stale_at)
assert %AuditEvent{event_type: "repository.git.write", repository_id: repository_id} =
Repo.get_by(AuditEvent,
event_type: "repository.git.write",
repository_id: repository.id
)
assert repository_id == repository.id
# A second clone sees the commit.
verify = seed_clone!(base, url)
assert File.read!(Path.join(verify, "hello.txt")) == "hello forge\n"
end
test "a second push appends to the WAL and receipts stay ordered", %{
base: base,
repository: repository,
url: url
} do
work = seed_clone!(base, url)
commit_and_push!(work, "a.txt", "one\n", "one")
commit_and_push!(work, "b.txt", "two\n", "two")
assert {:ok, _generation, index} = WAL.read_index(repository.storage_key)
assert length(WAL.entries(index)) == 2
assert [%{wal_seq: 1}, %{wal_seq: 0}] = Forge.recent_pushes(repository.storage_key)
end
test "cache loss: the bare repo re-materializes from the WAL", %{
base: base,
repository: repository,
url: url
} do
work = seed_clone!(base, url)
commit_and_push!(work, "keep.txt", "durable\n", "durable commit")
refs_before = Repos.refs(repository.storage_key)
# Kill the cache entirely.
File.rm_rf!(Repos.bare_path(repository.storage_key))
refute File.exists?(Repos.bare_path(repository.storage_key))
# A fresh clone triggers materialization and sees identical state.
verify = seed_clone!(base, url)
assert File.read!(Path.join(verify, "keep.txt")) == "durable\n"
assert Repos.refs(repository.storage_key) == refs_before
end
test "an unavailable cache returns 503 instead of a false repository 404", %{
repository: repository,
token: token
} do
entry = %{
"seq" => 0,
"object" => "entries/missing",
"format" => "git_bundle",
"refs" => %{"refs/heads/main" => String.duplicate("a", 40)},
"principal" => "test",
"pushed_at" => DateTime.to_iso8601(DateTime.utc_now())
}
{:ok, _generation} =
WAL.cas_index(
repository.storage_key,
:none,
WAL.append_entry(WAL.new_index(), entry)
)
authorization = "Basic " <> Base.encode64("x:#{token}")
response =
:get
|> Plug.Test.conn("/git-http-owner/demo.git/info/refs?service=git-upload-pack")
|> Plug.Conn.put_req_header("authorization", authorization)
|> TestPipeline.call([])
assert response.status == 503
assert response.resp_body == "repository cache unavailable; retry"
refute CacheReadiness.ready?()
end
test "unauthenticated and wrong-token pushes are refused", %{
base: base,
port: port,
repository: repository,
url: url
} do
work = seed_clone!(base, url)
File.write!(Path.join(work, "no.txt"), "no\n")
sh!(work, "git", ["add", "."])
sh!(work, "git", ["commit", "-m", "unauthorized"])
for bad_url <- [
"http://127.0.0.1:#{port}/git-http-owner/demo.git",
"http://x:wrong-token@127.0.0.1:#{port}/git-http-owner/demo.git"
] do
{output, status} =
System.cmd("git", ["-c", "credential.helper=", "push", bad_url, "HEAD:main"],
cd: work,
stderr_to_stdout: true,
env: [{"GIT_TERMINAL_PROMPT", "0"}]
)
assert status != 0
assert output =~ "401" or output =~ "Authentication" or
output =~ "terminal prompts disabled"
end
# Nothing leaked into the WAL.
case WAL.read_index(repository.storage_key) do
{:error, :not_found} -> :ok
{:ok, _generation, index} -> assert WAL.entries(index) == []
end
end
test "unknown repositories are refused", %{port: port, token: token} do
{output, status} =
System.cmd(
"git",
[
"-c",
"credential.helper=",
"clone",
"http://x:#{token}@127.0.0.1:#{port}/git-http-owner/not-allowed.git"
],
cd: System.tmp_dir!(),
stderr_to_stdout: true
)
assert status != 0
assert output =~ "404" or output =~ "not found" or output =~ "unknown"
end
test "the operator can access a configured UUID-backed repository", %{
base: base,
port: port,
repository: repository
} do
operator_token = "forge_operator_test_token_0123456789"
previous_token = Application.get_env(:openagents, :forge_operator_token)
previous_repos = Application.get_env(:openagents, :forge_repos)
Application.put_env(:openagents, :forge_operator_token, operator_token)
Application.put_env(:openagents, :forge_repos, [repository.name])
on_exit(fn ->
Application.put_env(:openagents, :forge_operator_token, previous_token)
Application.put_env(:openagents, :forge_repos, previous_repos)
end)
url =
"http://x:#{operator_token}@127.0.0.1:#{port}/git-http-owner/demo.git"
work = seed_clone!(base, url)
commit_and_push!(work, "operator.txt", "operator\n", "Operator commit")
assert [_receipt] = Forge.recent_pushes(repository.storage_key)
end
test "Git RPC reauthenticates a token after ref advertisement", %{
api_token: api_token,
token: token,
user: user
} do
authorization = "Basic " <> Base.encode64("x:#{token}")
advertised =
:get
|> Plug.Test.conn("/git-http-owner/demo.git/info/refs?service=git-upload-pack")
|> Plug.Conn.put_req_header("authorization", authorization)
|> TestPipeline.call([])
assert advertised.status == 200
assert {:ok, _revoked} = OpenAgents.ApiTokens.revoke(user, api_token.id)
rpc =
:post
|> Plug.Test.conn("/git-http-owner/demo.git/git-upload-pack", "")
|> Plug.Conn.put_req_header("authorization", authorization)
|> TestPipeline.call([])
assert rpc.status == 401
assert Plug.Conn.get_resp_header(rpc, "www-authenticate") != []
end
test "public repositories clone anonymously and private repositories issue a challenge", %{
base: base,
port: port,
repository: repository
} do
anonymous_url = "http://127.0.0.1:#{port}/git-http-owner/demo.git"
{private_output, private_status} =
System.cmd(
"git",
["-c", "credential.helper=", "clone", anonymous_url, Path.join(base, "private")],
stderr_to_stdout: true,
env: [{"GIT_TERMINAL_PROMPT", "0"}]
)
assert private_status != 0
assert private_output =~ "Authentication" or private_output =~ "terminal prompts disabled"
repository
|> Ecto.Changeset.change(visibility: "public")
|> OpenAgents.Repo.update!()
public_clone = Path.join(base, "public")
sh!(base, "git", ["clone", anonymous_url, public_clone])
assert File.dir?(Path.join(public_clone, ".git"))
end
test "a viewer can clone a private repository but cannot push", %{
base: base,
port: port,
repository: repository,
url: owner_url
} do
owner_clone = seed_clone!(base, owner_url)
commit_and_push!(owner_clone, "owner.txt", "owner\n", "Owner commit")
viewer = repository_user_fixture("git-http-viewer")
{:ok, _membership} = OpenAgents.Repositories.add_member(repository, viewer, "viewer")
{:ok, _api_token, viewer_token} =
OpenAgents.ApiTokens.create(viewer, %{
name: "Viewer Git HTTP test",
scopes: ["forge:write"],
lifetime_days: 1
})
viewer_url =
"http://x:#{viewer_token}@127.0.0.1:#{port}/git-http-owner/demo.git"
viewer_clone = seed_clone!(base, viewer_url)
File.write!(Path.join(viewer_clone, "viewer.txt"), "viewer\n")
sh!(viewer_clone, "git", ["add", "viewer.txt"])
sh!(viewer_clone, "git", ["commit", "-m", "Viewer commit"])
{output, status} =
System.cmd("git", ["-c", "credential.helper=", "push", "origin", "HEAD:main"],
cd: viewer_clone,
stderr_to_stdout: true
)
assert status != 0
assert output =~ "403" or output =~ "read only" or output =~ "unable to access"
end
test "a paired machine requires an explicit operation-scoped repository grant", %{
base: base,
port: port,
repository: repository,
user: user
} do
{:ok, pairing} =
Machines.start_pairing(%{
"name" => "git-machine",
"tier" => "probe",
"platform" => "linux",
"agent_version" => "0.1.0",
"roots" => []
})
assert {:ok, machine} = Machines.approve_pairing(user, pairing.code)
assert {:ok, %{token: machine_token}} =
Machines.claim_pairing(pairing.pairing.id, pairing.poll_secret)
machine_url =
"http://x:#{machine_token}@127.0.0.1:#{port}/git-http-owner/demo.git"
{ungranted_output, ungranted_status} =
System.cmd(
"git",
["-c", "credential.helper=", "clone", machine_url, Path.join(base, "machine-ungranted")],
stderr_to_stdout: true,
env: [{"GIT_TERMINAL_PROMPT", "0"}]
)
assert ungranted_status != 0
assert ungranted_output =~ "404" or ungranted_output =~ "not found"
# Through the surface the owner actually has, not through the context
# function that used to have no caller in `lib/` (#182).
assert {:ok, _grant} =
Repositories.grant_machine_access(user, machine.id, repository.id, ["read"])
machine_clone = seed_clone!(base, machine_url)
File.write!(Path.join(machine_clone, "machine.txt"), "machine\n")
sh!(machine_clone, "git", ["add", "machine.txt"])
sh!(machine_clone, "git", ["commit", "-m", "Machine commit"])
{_output, read_only_status} =
System.cmd("git", ["-c", "credential.helper=", "push", "origin", "HEAD:main"],
cd: machine_clone,
stderr_to_stdout: true
)
assert read_only_status != 0
assert {:ok, _grant} =
Repositories.grant_machine_access(user, machine.id, repository.id, ["read", "write"])
sh!(machine_clone, "git", ["push", "origin", "HEAD:main"])
end
test "an upstream mirror serves a complete clone and refuses every push", %{
base: base,
port: port,
token: token,
user: user
} do
source = Path.join(base, "walgit-source")
File.mkdir_p!(source)
sh!(source, "git", ["init", "--initial-branch=main"])
sh!(source, "git", ["config", "user.email", "test@example.com"])
sh!(source, "git", ["config", "user.name", "Upstream"])
Enum.each(1..3, fn index ->
File.write!(Path.join(source, "README.md"), "revision #{index}\n")
sh!(source, "git", ["add", "README.md"])
sh!(source, "git", ["commit", "-m", "Revision #{index}"])
end)
head = source |> sh!("git", ["rev-parse", "HEAD"]) |> String.trim()
refs = %{"refs/heads/main" => head}
{:ok, mirror, _import, :created} =
Repositories.create_user_mirror(
user,
%{
source_repository_id: 909,
source_owner_id: 777_777,
source_full_name: "tobi/walgit",
source_default_branch: "main",
source_ref_digest: ref_digest(source, refs),
source_head_sha: head,
source_refs: refs,
source_uses_lfs: false,
source_public: true,
source_license: "MIT"
},
%{name: "walgit", visibility: "public", default_branch: "main"},
"git-http-mirror-key"
)
:ok = OpenAgents.Repositories.Importer.import(mirror, source_url: source)
mirror
|> Ecto.Changeset.change(lifecycle_state: "ready", ready_at: DateTime.utc_now())
|> Repo.update!()
mirror_url = "http://x:#{token}@127.0.0.1:#{port}/git-http-owner/walgit.git"
# A mirror is still a forge repository: EXIT-004 holds for it. The clone
# walks to the root because the copy was taken whole, so no boundary can
# abort the transfer.
clone = Path.join(base, "walgit-clone")
sh!(base, "git", ["clone", mirror_url, clone])
assert String.trim(sh!(clone, "git", ["rev-list", "--count", "HEAD"])) == "3"
# `sh!/3` flunks on a nonzero exit, so this asserts fsck succeeded; the
# refutations catch the dangling and missing objects fsck reports while
# still exiting zero.
fsck = sh!(clone, "git", ["fsck", "--full"])
refute fsck =~ "missing"
refute fsck =~ "broken"
sh!(clone, "git", ["config", "user.email", "test@example.com"])
sh!(clone, "git", ["config", "user.name", "Forge Test"])
File.write!(Path.join(clone, "local.txt"), "local\n")
sh!(clone, "git", ["add", "local.txt"])
sh!(clone, "git", ["commit", "-m", "Local commit"])
{output, status} =
System.cmd("git", ["-c", "credential.helper=", "push", "origin", "HEAD:main"],
cd: clone,
stderr_to_stdout: true
)
assert status != 0
assert output =~ "403"
# The git client stops at the ref advertisement, so a refusal that lived
# only there would look complete while `POST /git-receive-pack` stayed
# open. Both are asserted, against the same account whose token clones the
# repository fine.
advertisement =
Req.get!(
"http://127.0.0.1:#{port}/git-http-owner/walgit.git/info/refs?service=git-receive-pack",
auth: {:basic, "x:#{token}"},
retry: false
)
assert advertisement.status == 403
assert advertisement.body =~ "upstream mirror of https://github.com/tobi/walgit"
assert advertisement.body =~ "accepts no pushes"
receive_pack =
Req.post!(
"http://127.0.0.1:#{port}/git-http-owner/walgit.git/git-receive-pack",
auth: {:basic, "x:#{token}"},
headers: [{"content-type", "application/x-git-receive-pack-request"}],
body: "0000",
retry: false
)
assert receive_pack.status == 403
assert receive_pack.body =~ "upstream mirror of https://github.com/tobi/walgit"
# Nothing moved: the mirror still holds exactly what the upstream had.
assert Repos.refs(mirror.storage_key) == refs
end
defp ref_digest(source, refs) do
refs
|> Enum.sort_by(&elem(&1, 0))
|> Enum.map_join("\n", fn {name, sha} ->
object_type = source |> sh!("git", ["cat-file", "-t", sha]) |> String.trim()
Enum.join([name, object_type, sha], "\0")
end)
|> then(&:crypto.hash(:sha256, &1))
|> Base.encode16(case: :lower)
end
test "the legacy initial repository path remains available", %{
base: base,
port: port,
token: token
} do
legacy_url = "http://x:#{token}@127.0.0.1:#{port}/openagents.com.git"
legacy_clone = Path.join(base, "legacy")
sh!(base, "git", ["clone", legacy_url, legacy_clone])
assert File.dir?(Path.join(legacy_clone, ".git"))
end
test "a failed WAL persist rolls refs back and the push is not acked", %{
base: base,
repository: repository,
url: url
} do
work = seed_clone!(base, url)
commit_and_push!(work, "ok.txt", "fine\n", "fine")
refs_before = Repos.refs(repository.storage_key)
# Break the WAL (unwritable dir) — the next push must not ack.
wal_dir = Application.get_env(:openagents, :forge_wal_dir)
File.chmod!(Path.join(wal_dir, repository.storage_key), 0o500)
File.write!(Path.join(work, "lost.txt"), "must not land\n")
sh!(work, "git", ["add", "."])
sh!(work, "git", ["commit", "-m", "must not land"])
{output, status} =
System.cmd("git", ["-c", "credential.helper=", "push", "origin", "HEAD:main"],
cd: work,
stderr_to_stdout: true
)
File.chmod!(Path.join(wal_dir, repository.storage_key), 0o700)
assert status != 0, "push must fail when the WAL cannot persist: #{output}"
# Local refs rolled back — cache never ahead of authority.
assert Repos.refs(repository.storage_key) == refs_before
assert {:ok, _generation, index} = WAL.read_index(repository.storage_key)
assert length(WAL.entries(index)) == 1
end
# The side-band line #167 returns to the pusher. `git` treats an
# unparseable report-status as a failed push, so the shapes this function
# must refuse to touch are asserted directly. Every client in the tests
# above negotiates `side-band-64k`, so a bare report-status never reaches
# the function through them and would otherwise go unproven.
describe "the WAL receipt side-band line" do
alias OpenAgents.Forge.GitHTTP
@report_status GitHTTP.pkt_line("unpack ok\n") <>
GitHTTP.pkt_line("ok refs/heads/main\n") <> "0000"
test "a bare report-status is returned byte for byte" do
assert GitHTTP.append_side_band(@report_status, "openagents wal-receipt seq=1") ==
@report_status
end
test "a side-band stream gains one band-2 packet before the final flush" do
framed = GitHTTP.pkt_line(<<1>> <> @report_status) <> "0000"
annotated = GitHTTP.append_side_band(framed, "openagents wal-receipt seq=1 link=abc")
assert annotated ==
GitHTTP.pkt_line(<<1>> <> @report_status) <>
GitHTTP.pkt_line(<<2>> <> "openagents wal-receipt seq=1 link=abc\n") <> "0000"
# The report-status the client parses is untouched.
assert String.starts_with?(annotated, GitHTTP.pkt_line(<<1>> <> @report_status))
end
test "a stream that is not a pkt-line stream, or does not end in a flush, is untouched" do
for output <- ["", "0000", "not a packet stream", GitHTTP.pkt_line(<<1>> <> "ok\n")] do
assert GitHTTP.append_side_band(output, "openagents wal-receipt seq=1") == output
end
end
test "a message with control characters becomes one printable line" do
framed = GitHTTP.pkt_line(<<1>> <> @report_status) <> "0000"
annotated = GitHTTP.append_side_band(framed, "seq=1\nlink=abc\r\n\tmore")
assert annotated =~ "seq=1 link=abc more"
refute String.contains?(annotated, "\r")
# One packet, one trailing newline: the message cannot smuggle a second
# line into the stream.
assert annotated
|> String.split(<<2>>)
|> List.last()
|> String.trim_trailing("0000") =~ ~r/\A[^\n]+\n\z/
end
end
end