u78649634_default/smart-org
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(delivery): git workspaces + SSRF host check (v2 phase A3-2)

Browse Source 
The platform's side of D6: git runs on the host, in a gitdir that is never
mounted into a container, so repository credentials never reach the model.

- adapters/git_workspace.py: LocalGitWorkspaceAdapter implementing
  RepoWorkspacePort. Bare shared gitdir + a dev work tree mounted at
  /workspace + one linked worktree per task under work/.worktrees/<id>
  (excluded via the gitdir's info/exclude, not the project's .gitignore).
  Explicit --git-dir/--work-tree, core.hooksPath=/dev/null, fsmonitor off,
  quotePath off, GIT_CONFIG_NOSYSTEM, a private HOME per project, prompts off,
  one asyncio.Lock per project. Token via GIT_ASKPASS reading a 0600 file; SSH
  key as a 0600 temp file with a pinned known_hosts. Everything deleted in a
  finally; only a *path* ever travels through the environment.
- The bug this phase existed to find: --git-dir=<shared> does not address a
  linked worktree, so `git commit` in a task worktree committed straight to
  `dev`. Operations now address <gitdir>/worktrees/<task_id> explicitly — and
  never read the worktree's .git file, which a repository can rewrite.
- domain/platform/rules.py: is_blocked_address (pure) + assert_remote_host_allowed
  (resolves, refuses if any address is loopback/private/link-local/metadata).
  Shared by the git remote test and, later, every deploy target.
- ProjectExternalGit domain model + repository; container_delivery wires the
  real adapter behind GIT_WORKSPACE=local with a resolver that joins the
  project, its repo and its (decrypted) credentials.

Tests: 394 API (+48) against real git and a local bare repo; ruff and mypy clean.

Report: docs/plan/reports/agent-3-phase-2.md
This commit is contained in:
mjalmoudy94
2026-10-14 00:18:13 +03:30
parent a0c8473eb8
commit d41e00ad10
10 changed files with 2242 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/tests/test_git_workspace.py
+693
View File
@@ -0,0 +1,693 @@
"""§v2-A3.2 — the platform's git adapter, exercised against real git.
These tests run a real ``git`` binary against a local bare repository standing in
for the remote. Nothing is faked except the *remote's location*: the flags, the
locking, the worktree dance and the credential plumbing are the real thing,
because those are precisely the parts that cannot be verified by a mock.
``protocol.file.allow=user`` is passed explicitly. The production default is
``never`` — a repository must not be able to make git read a local path through a
submodule or ``--upload-pack`` — but a local bare repo *is* a local path, so the
hardening has to be relaxed for the test to be able to run at all. That is why
the setting exists rather than being hard-coded.
"""
from __future__ import annotations
import asyncio
import os
import pytest
from app.core.errors import NotFoundError
from app.domain.platform.rules import RuleViolation, assert_remote_host_allowed, is_blocked_address
from app.infrastructure.adapters.git_workspace import (
LocalGitWorkspaceAdapter,
ProjectGitSource,
)
from app.infrastructure.adapters.process import ProcessResult, run_process
# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------
async def _git(args: list[str], *, cwd: str, ok: bool = True) -> ProcessResult:
result = await run_process(
["git", *args],
cwd=cwd,
timeout_seconds=60,
env={
"GIT_CONFIG_NOSYSTEM": "1",
"GIT_TERMINAL_PROMPT": "0",
"GIT_AUTHOR_NAME": "t",
"GIT_AUTHOR_EMAIL": "t@example.com",
"GIT_COMMITTER_NAME": "t",
"GIT_COMMITTER_EMAIL": "t@example.com",
},
)
if ok:
assert result.ok, f"git {' '.join(args)} failed: {result.stderr}"
return result
async def _make_remote(tmp_path) -> str:
"""A local bare repository with one commit on ``main``; returns its path."""
remote = tmp_path / "remote.git"
seed = tmp_path / "seed"
await _git(["init", "--quiet", "--bare", str(remote)], cwd=str(tmp_path))
await _git(["init", "--quiet", str(seed)], cwd=str(tmp_path))
(seed / "README.md").write_text("hello\n")
await _git(["add", "-A"], cwd=str(seed))
await _git(["commit", "--quiet", "-m", "init"], cwd=str(seed))
await _git(["branch", "-M", "main"], cwd=str(seed))
await _git(["remote", "add", "origin", str(remote)], cwd=str(seed))
await _git(["push", "--quiet", "origin", "main"], cwd=str(seed))
return str(remote)
def _resolver(clone_url: str, *, auth_type: str = "none", secret: str | None = None,
username: str | None = None):
async def resolve(project_id: str) -> ProjectGitSource | None:
if project_id == "missing":
return None
return ProjectGitSource(
project_id=project_id,
organization_id="org1",
clone_url=clone_url,
default_branch="main",
auth_type=auth_type,
username=username,
secret=secret,
)
return resolve
def _adapter(tmp_path, clone_url: str, **kwargs) -> LocalGitWorkspaceAdapter:
kwargs.setdefault("allow_file_protocol", True)
kwargs.setdefault("runner", run_process)
return LocalGitWorkspaceAdapter(
resolver=_resolver(clone_url, **kwargs.pop("credentials", {})),
root=str(tmp_path / "root"),
**kwargs,
)
# ---------------------------------------------------------------------------
# ensure_workspace
# ---------------------------------------------------------------------------
async def test_ensure_workspace_clones_and_creates_dev(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
ref = await adapter.ensure_workspace(project_id="proj1")
assert ref.default_branch == "dev"
assert os.path.isdir(ref.host_path), "the work tree must exist for the sandbox mount"
assert (tmp_path / "root" / "org1" / "proj1" / "work" / "README.md").is_file()
assert await adapter.head(project_id="proj1", branch="dev") is not None
# `dev` was pushed to the platform remote because it did not exist there.
assert await adapter.head(project_id="proj1", branch="dev") is not None
async def test_ensure_workspace_is_idempotent(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
first = await adapter.ensure_workspace(project_id="proj1")
(tmp_path / "root" / "org1" / "proj1" / "work" / "new.txt").write_text("x")
second = await adapter.ensure_workspace(project_id="proj1")
assert first.host_path == second.host_path
assert (tmp_path / "root" / "org1" / "proj1" / "work" / "new.txt").is_file()
async def test_ensure_workspace_rejects_an_unknown_project(tmp_path):
adapter = _adapter(tmp_path, str(tmp_path / "nothing.git"))
with pytest.raises(NotFoundError):
await adapter.ensure_workspace(project_id="missing")
async def test_gitdir_is_not_inside_the_work_tree(tmp_path):
"""The git dir must never be reachable from the sandbox mount (D6)."""
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
ref = await adapter.ensure_workspace(project_id="proj1")
gitdir = tmp_path / "root" / "org1" / "proj1" / "git"
assert gitdir.is_dir()
assert not os.path.exists(os.path.join(ref.host_path, ".git"))
assert not str(gitdir).startswith(ref.host_path + os.sep)
# ---------------------------------------------------------------------------
# worktrees
# ---------------------------------------------------------------------------
async def test_task_worktree_is_created_and_removed(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
assert worktree.branch == "task/t1"
assert worktree.sandbox_path == "/workspace/.worktrees/t1"
assert os.path.isdir(worktree.host_path)
await adapter.remove_worktree(worktree=worktree)
assert not os.path.isdir(worktree.host_path)
async def test_task_worktree_is_invisible_in_the_main_work_tree(tmp_path):
"""`info/exclude` keeps .worktrees out of `status` without touching .gitignore."""
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
ref = await adapter.ensure_workspace(project_id="proj1")
await adapter.create_task_worktree(project_id="proj1", task_id="t1")
result = await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"--work-tree", ref.host_path,
"status", "--porcelain=v1", "-uall"],
cwd=ref.host_path,
)
assert result.stdout.strip() == "", "the worktree dir must not show up as untracked"
assert not os.path.exists(os.path.join(ref.host_path, ".gitignore"))
async def test_create_task_worktree_is_idempotent(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
first = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
second = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
assert first.host_path == second.host_path
# ---------------------------------------------------------------------------
# diff / commit
# ---------------------------------------------------------------------------
async def test_diff_reports_a_new_directory_thanks_to_uall(tmp_path):
"""`git status` without ``-uall`` hides new directories — a real agybot bug."""
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
(tmp_path / "root" / "org1" / "proj1" / "work" / ".worktrees" / "t1" / "src").mkdir()
(
tmp_path / "root" / "org1" / "proj1" / "work" / ".worktrees" / "t1" / "src" / "a.py"
).write_text("print('hi')\n")
diff = await adapter.diff(worktree=worktree)
assert diff.changed_files == ["src/a.py"]
assert diff.is_empty is False
async def test_diff_detects_an_escaping_symlink(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
outside = tmp_path / "outside.txt"
outside.write_text("secret")
os.symlink(outside, os.path.join(worktree.host_path, "escape.txt"))
diff = await adapter.diff(worktree=worktree)
assert diff.escaping_symlinks == ["escape.txt"]
async def test_commit_returns_none_when_there_is_nothing_to_commit(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
assert await adapter.commit(worktree=worktree, message="empty", author="AI") is None
async def test_commit_records_the_author_and_returns_a_sha(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
target = os.path.join(worktree.host_path, "feature.py")
with open(target, "w", encoding="utf-8") as handle:
handle.write("x = 1\n")
sha = await adapter.commit(worktree=worktree, message="task/t1: add feature", author="Alice")
assert sha and len(sha) == 40
shared_gitdir = tmp_path / "root" / "org1" / "proj1" / "git"
log = await _git(
["--git-dir", str(shared_gitdir), "log", "-1", "--format=%an <%ae>|%s", "task/t1"],
cwd=str(tmp_path),
)
assert log.stdout.strip() == "Alice <alice@smart-org.local>|task/t1: add feature"
# ...and specifically to `task/t1`, not to `dev`. A shared gitdir makes this
# the easiest thing in the world to get wrong.
task_head = await _git(
["--git-dir", str(shared_gitdir), "rev-parse", "task/t1"], cwd=str(tmp_path)
)
assert task_head.stdout.strip() == sha
dev_head = await _git(
["--git-dir", str(shared_gitdir), "rev-parse", "dev"], cwd=str(tmp_path)
)
assert dev_head.stdout.strip() != sha, "the commit must not land on dev directly"
# ---------------------------------------------------------------------------
# merge / push / head
# ---------------------------------------------------------------------------
async def test_merge_into_dev_and_push(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
with open(os.path.join(worktree.host_path, "a.txt"), "w", encoding="utf-8") as handle:
handle.write("a\n")
sha = await adapter.commit(worktree=worktree, message="task/t1: a", author="AI")
merged = await adapter.merge_into(project_id="proj1", branch="task/t1", target="dev")
assert merged.ok is True, merged.message
# `--no-ff` means dev gets a merge *commit*, not the task commit itself, so
# the task sha is a parent of dev's new head rather than dev's head.
dev_head = await adapter.head(project_id="proj1", branch="dev")
assert dev_head == merged.sha
assert dev_head != sha
ancestry = await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"rev-list", "--parents", "-1", "dev"],
cwd=str(tmp_path),
)
assert sha in ancestry.stdout.split()
pushed = await adapter.push(project_id="proj1", branch="dev")
assert pushed.ok is True, pushed.message
# The remote really received it.
ls = await _git(["--git-dir", remote, "rev-parse", "dev"], cwd=str(tmp_path))
assert ls.stdout.strip() == dev_head
async def test_merge_conflict_leaves_a_clean_tree(tmp_path):
"""A half-merge would break every later operation (agybot bug H3).
The task branch has to fork *before* ``dev`` moves, otherwise the merge is a
trivial fast-forward and there is nothing to conflict about.
"""
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
# 1. The task branches off dev as it stands.
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
with open(os.path.join(worktree.host_path, "README.md"), "w", encoding="utf-8") as handle:
handle.write("task version\n")
await adapter.commit(worktree=worktree, message="task/t1: edit", author="AI")
# 2. dev moves on, touching the same file.
main_work = tmp_path / "root" / "org1" / "proj1" / "work"
(main_work / "README.md").write_text("dev version\n")
await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"--work-tree", str(main_work), "add", "-A"],
cwd=str(main_work),
)
await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"--work-tree", str(main_work), "commit", "--quiet", "-m", "dev edit"],
cwd=str(main_work),
)
outcome = await adapter.merge_into(project_id="proj1", branch="task/t1", target="dev")
assert outcome.ok is False
assert outcome.conflict_files == ["README.md"]
# No merge in progress: the next status is clean.
status = await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"--work-tree", str(main_work), "status", "--porcelain=v1", "-uall"],
cwd=str(main_work),
)
assert status.stdout.strip() == ""
assert not os.path.exists(
tmp_path / "root" / "org1" / "proj1" / "git" / "MERGE_HEAD"
)
async def test_sync_with_base_reports_that_the_base_moved(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
# Nothing moved yet.
quiet = await adapter.sync_with_base(worktree=worktree)
assert quiet.ok is True and quiet.base_changed is False
# Move dev forward.
main_work = tmp_path / "root" / "org1" / "proj1" / "work"
(main_work / "other.txt").write_text("x\n")
await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"--work-tree", str(main_work), "add", "-A"],
cwd=str(main_work),
)
await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"--work-tree", str(main_work), "commit", "--quiet", "-m", "dev moved"],
cwd=str(main_work),
)
moved = await adapter.sync_with_base(worktree=worktree)
assert moved.ok is True
assert moved.base_changed is True
assert os.path.isfile(os.path.join(worktree.host_path, "other.txt"))
async def test_concurrent_merges_are_serialised(tmp_path):
"""Without the per-project lock two merges corrupt the index."""
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
for task in ("t1", "t2"):
worktree = await adapter.create_task_worktree(project_id="proj1", task_id=task)
with open(
os.path.join(worktree.host_path, f"{task}.txt"), "w", encoding="utf-8"
) as handle:
handle.write(f"{task}\n")
await adapter.commit(worktree=worktree, message=f"task/{task}", author="AI")
first, second = await asyncio.gather(
adapter.merge_into(project_id="proj1", branch="task/t1", target="dev"),
adapter.merge_into(project_id="proj1", branch="task/t2", target="dev"),
)
assert first.ok is True and second.ok is True
status = await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"--work-tree", str(tmp_path / "root" / "org1" / "proj1" / "work"),
"status", "--porcelain=v1", "-uall"],
cwd=str(tmp_path),
)
assert status.stdout.strip() == ""
async def test_push_reports_a_rejection_without_forcing(tmp_path):
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
# Someone else advances the remote's dev, so our push is no longer a
# fast-forward. `origin/dev` is what the clone checked out, not a fresh
# branch off the default one.
other = tmp_path / "other"
await _git(["clone", "--quiet", remote, str(other)], cwd=str(tmp_path))
await _git(["checkout", "--quiet", "-B", "dev", "origin/dev"], cwd=str(other))
(other / "divergent.txt").write_text("other\n")
await _git(["add", "-A"], cwd=str(other))
await _git(["commit", "--quiet", "-m", "divergent"], cwd=str(other))
await _git(["push", "--quiet", "origin", "dev"], cwd=str(other))
# Local dev moves a different way.
main_work = tmp_path / "root" / "org1" / "proj1" / "work"
(main_work / "mine.txt").write_text("mine\n")
await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"--work-tree", str(main_work), "add", "-A"],
cwd=str(main_work),
)
await _git(
["--git-dir", str(tmp_path / "root" / "org1" / "proj1" / "git"),
"--work-tree", str(main_work), "commit", "--quiet", "-m", "mine"],
cwd=str(main_work),
)
outcome = await adapter.push(project_id="proj1", branch="dev")
assert outcome.ok is False
assert outcome.rejected is True, outcome.message
# ---------------------------------------------------------------------------
# hardening
# ---------------------------------------------------------------------------
async def test_hooks_are_never_executed(tmp_path):
"""A repository may ship hooks; a hook is arbitrary host code execution."""
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
gitdir = tmp_path / "root" / "org1" / "proj1" / "git"
hooks = gitdir / "hooks"
hooks.mkdir(parents=True, exist_ok=True)
marker = tmp_path / "HOOK_RAN"
hook = hooks / "post-commit"
hook.write_text(f"#!/bin/sh\ntouch {marker}\n")
os.chmod(hook, 0o755)
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
with open(os.path.join(worktree.host_path, "a.txt"), "w", encoding="utf-8") as handle:
handle.write("a\n")
await adapter.commit(worktree=worktree, message="task/t1", author="AI")
assert not marker.exists(), "a post-commit hook must not run"
async def test_a_tampered_dot_git_file_is_ignored(tmp_path):
"""The `.git` pointer inside a worktree is attacker-controlled text."""
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
await adapter.ensure_workspace(project_id="proj1")
worktree = await adapter.create_task_worktree(project_id="proj1", task_id="t1")
decoy = tmp_path / "decoy"
await _git(["init", "--quiet", str(decoy)], cwd=str(tmp_path))
with open(os.path.join(worktree.host_path, ".git"), "w", encoding="utf-8") as handle:
handle.write(f"gitdir: {decoy}\n")
with open(os.path.join(worktree.host_path, "a.txt"), "w", encoding="utf-8") as handle:
handle.write("a\n")
sha = await adapter.commit(worktree=worktree, message="task/t1", author="AI")
assert sha is not None
# The commit landed on OUR task branch, not in the decoy.
shared_gitdir = tmp_path / "root" / "org1" / "proj1" / "git"
real = await _git(
["--git-dir", str(shared_gitdir), "rev-parse", "task/t1"], cwd=str(tmp_path)
)
assert real.stdout.strip() == sha
decoy_log = await _git(
["--git-dir", str(decoy), "log", "--oneline"], cwd=str(tmp_path), ok=False
)
assert decoy_log.exit_code != 0 or "task/t1" not in decoy_log.stdout
async def test_credentials_never_appear_in_argv(tmp_path):
"""argv is world-readable via `ps`; a secret must travel another way."""
secret = "ghp_thisisasecrettokenvalue0123456789"
seen: list[tuple[str, ...]] = []
async def recording_runner(argv, **kwargs): # noqa: ANN001, ANN003
seen.append(tuple(argv))
return await run_process(argv, **kwargs)
remote = await _make_remote(tmp_path)
adapter = LocalGitWorkspaceAdapter(
resolver=_resolver(remote, auth_type="token", secret=secret, username="git"),
root=str(tmp_path / "root"),
allow_file_protocol=True,
runner=recording_runner,
)
# A token against a local path fails, but the *attempt* is what we assert on.
await adapter.ensure_workspace(project_id="proj1")
assert seen, "the adapter must have invoked git"
for argv in seen:
joined = " ".join(argv)
assert secret not in joined, f"secret leaked into argv: {argv}"
async def test_the_credential_file_is_removed_after_the_call(tmp_path):
remote = await _make_remote(tmp_path)
adapter = LocalGitWorkspaceAdapter(
resolver=_resolver(remote, auth_type="token", secret="s3cret-token-value", username="git"),
root=str(tmp_path / "root"),
allow_file_protocol=True,
)
await adapter.ensure_workspace(project_id="proj1")
private = tmp_path / "root" / "org1" / "proj1" / "private"
assert private.is_dir()
assert not (private / "credentials").exists(), "the 0600 secret file must be deleted"
assert not (private / "id_key").exists()
async def test_the_askpass_helper_is_private_and_holds_no_secret(tmp_path):
remote = await _make_remote(tmp_path)
adapter = LocalGitWorkspaceAdapter(
resolver=_resolver(remote, auth_type="token", secret="s3cret-token-value", username="git"),
root=str(tmp_path / "root"),
allow_file_protocol=True,
)
await adapter.ensure_workspace(project_id="proj1")
helper = tmp_path / "root" / "org1" / "proj1" / "private" / "askpass.sh"
assert helper.is_file()
assert oct(helper.stat().st_mode)[-3:] == "700"
assert "s3cret-token-value" not in helper.read_text()
# ---------------------------------------------------------------------------
# SSRF: the shared host check
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"address",
[
"127.0.0.1",
"10.0.0.1",
"172.16.0.1",
"192.168.1.1",
"169.254.169.254",
"::1",
"fd00::1",
"fe80::1",
"0.0.0.0",
],
)
def test_is_blocked_address_covers_the_dangerous_ranges(address: str):
assert is_blocked_address(address) is True
@pytest.mark.parametrize("address", ["8.8.8.8", "1.1.1.1", "2606:4700::1111"])
def test_is_blocked_address_allows_public_addresses(address: str):
assert is_blocked_address(address) is False
@pytest.mark.parametrize(
"url",
[
"http://169.254.169.254/latest/meta-data/",
"http://127.0.0.1:6379/",
"http://localhost/repo.git",
"https://10.0.0.5/repo.git",
"http://[::1]/repo.git",
"https://user:pass@example.com/repo.git",
"file:///etc/passwd",
"https://example.com:6379/repo.git",
],
)
def test_assert_remote_host_allowed_refuses_dangerous_targets(url: str):
with pytest.raises(RuleViolation):
assert_remote_host_allowed(url, resolver=lambda *a, **k: [])
def test_assert_remote_host_allowed_accepts_a_public_host():
def resolver(host, port, **kwargs): # noqa: ANN001, ANN003
return [(2, 1, 6, "", ("93.184.216.34", port))]
host, port = assert_remote_host_allowed(
"https://git.example.com/team/app.git", resolver=resolver
)
assert host == "git.example.com"
assert port is None
def test_assert_remote_host_allowed_refuses_a_name_that_resolves_private():
"""A public-looking name can resolve into the private network."""
def resolver(host, port, **kwargs): # noqa: ANN001, ANN003
return [(2, 1, 6, "", ("10.1.2.3", port))]
with pytest.raises(RuleViolation):
assert_remote_host_allowed("https://internal.example.com/repo.git", resolver=resolver)
def test_assert_remote_host_allowed_can_be_told_to_allow_private():
host, _ = assert_remote_host_allowed(
"http://127.0.0.1:8080/repo.git", allow_private=True
)
assert host == "127.0.0.1"
def test_assert_remote_host_allowed_accepts_scp_style_ssh():
def resolver(host, port, **kwargs): # noqa: ANN001, ANN003
return [(2, 1, 6, "", ("93.184.216.34", port))]
host, _ = assert_remote_host_allowed("git@git.example.com:team/app.git", resolver=resolver)
assert host == "git.example.com"
async def test_test_remote_refuses_a_private_host(tmp_path):
adapter = _adapter(tmp_path, "https://example.com/x.git")
from app.application.ports.repo_workspace import RemoteCredentials
check = await adapter.test_remote(
url="http://169.254.169.254/repo.git", credentials=RemoteCredentials()
)
assert check.ok is False
assert "داخلی" in check.message
async def test_test_remote_lists_branches_of_a_local_repo(tmp_path, monkeypatch):
"""The ls-remote path, with the SSRF check stubbed out.
The check itself is covered by the tests above; here the subject is the
parsing of `git ls-remote --heads` and the branch list it produces. A local
bare repo is used as the remote because the suite must not need a network.
"""
import app.infrastructure.adapters.git_workspace as module
from app.application.ports.repo_workspace import RemoteCredentials
monkeypatch.setattr(module, "assert_remote_host_allowed", lambda *a, **k: ("local", None))
remote = await _make_remote(tmp_path)
adapter = _adapter(tmp_path, remote)
check = await adapter.test_remote(
url=f"file://{remote}", credentials=RemoteCredentials()
)
assert check.ok is True, check.message
assert check.branches == ["main"]
assert check.default_branch == "main"
async def test_test_remote_reports_a_missing_repository(tmp_path, monkeypatch):
import app.infrastructure.adapters.git_workspace as module
from app.application.ports.repo_workspace import RemoteCredentials
monkeypatch.setattr(module, "assert_remote_host_allowed", lambda *a, **k: ("local", None))
adapter = _adapter(tmp_path, "https://example.com/x.git")
check = await adapter.test_remote(
url=f"file://{tmp_path}/does-not-exist.git", credentials=RemoteCredentials()
)
assert check.ok is False
assert check.message