diff --git a/pygitweb/main.py b/pygitweb/main.py index 1173f06..33a9336 100644 --- a/pygitweb/main.py +++ b/pygitweb/main.py @@ -100,6 +100,7 @@ from pygitweb.projects import git_get_project_owner, git_get_projects_list from pygitweb.sessions import clear_all_sessions from pygitweb.settings import router as settings_router from pygitweb.shutdown import begin_shutdown, install_graceful_shutdown_wakeup +from pygitweb.smart_http import http_router from pygitweb.tasks import ( board_router, comment_router, @@ -203,6 +204,7 @@ app.include_router(board_router, prefix="/board") app.include_router(task_router, prefix="/tasks") app.include_router(comment_router, prefix="/comments") app.include_router(merge_router, prefix="/mr") +app.include_router(http_router) @app.middleware("http") diff --git a/pygitweb/smart_http.py b/pygitweb/smart_http.py new file mode 100644 index 0000000..a0821fd --- /dev/null +++ b/pygitweb/smart_http.py @@ -0,0 +1,203 @@ +"""Git Smart HTTP transport routes (clone/fetch/push over HTTP).""" + +from __future__ import annotations + +import gzip +import os +import subprocess +from pathlib import Path +from typing import Annotated, Literal + +import pygit2 +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, status +from fastapi.responses import Response + +from pygitweb.auth import access_token_from_request, ensure_active_user_if_auth_enabled +from pygitweb.config import settings +from pygitweb.dependencies import project_visible_in_list, require_valid_project +from pygitweb.validation import is_valid_project + +SmartHttpService = Literal["git-upload-pack", "git-receive-pack"] + +http_router = APIRouter(prefix="/http", tags=["smart_http"]) + +_SERVICE_CONTENT_TYPES: dict[SmartHttpService, tuple[str, str]] = { + "git-upload-pack": ( + "application/x-git-upload-pack-advertisement", + "application/x-git-upload-pack-result", + ), + "git-receive-pack": ( + "application/x-git-receive-pack-advertisement", + "application/x-git-receive-pack-result", + ), +} + + +def _decode_request_body(body: bytes, content_encoding: str | None) -> bytes: + if not content_encoding or not content_encoding.strip(): + return body + encoding = content_encoding.strip().lower() + if encoding in ("identity", "none"): + return body + if encoding == "gzip": + try: + return gzip.decompress(body) + except OSError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid gzip body") from exc + raise HTTPException( + status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, + detail=f"Unsupported Content-Encoding: {content_encoding}", + ) + + +def _pkt_line(data: bytes) -> bytes: + length = len(data) + 4 + return f"{length:04x}".encode("ascii") + data + + +def _pkt_flush() -> bytes: + return b"0000" + + +def _is_valid_http_project(project: str) -> bool: + return is_valid_project( + project, + settings.PROJECTROOT, + settings.EXPORT_OK, + settings.STRICT_EXPORT, + project_visible_in_list, + ) + + +def require_http_project(project: str) -> str: + candidate = project.removesuffix(".git") + if candidate != project and _is_valid_http_project(candidate): + return candidate + return require_valid_project(project) + + +ValidatedHttpProject = Annotated[str, Depends(require_http_project)] + + +def _repo_cwd(project: str) -> Path: + full = os.path.join(settings.PROJECTROOT, project) + try: + git_dir = pygit2.discover_repository(full) + except (KeyError, pygit2.GitError, OSError) as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such project") from exc + if not git_dir: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such project") + repo = pygit2.Repository(git_dir) + if repo.is_bare: + return Path(git_dir) + worktree = repo.workdir + if worktree is None: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Repository has no worktree") + return Path(worktree) + + +def _run_git_service( + service: SmartHttpService, + project: str, + *, + advertise: bool, + body: bytes | None = None, + git_protocol: str | None = None, +) -> bytes: + cwd = _repo_cwd(project) + args = [settings.GIT, service.removeprefix("git-"), "--stateless-rpc" if not advertise else "--advertise-refs", "."] + env = os.environ.copy() + if git_protocol: + env["GIT_PROTOCOL"] = git_protocol + try: + completed = subprocess.run( + args, + cwd=str(cwd), + input=body, + capture_output=True, + check=False, + env=env, + ) + except OSError as exc: + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc + if completed.returncode != 0: + detail = completed.stderr.decode("utf-8", errors="replace").strip() or f"git {service} failed" + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=detail) + return completed.stdout + + +def _service_advertisement(service: SmartHttpService, project: str) -> bytes: + refs = _run_git_service(service, project, advertise=True) + announce = _pkt_line(f"# service={service}\n".encode("ascii")) + return announce + _pkt_flush() + refs + + +def _smart_http_cache_headers() -> dict[str, str]: + return { + "Cache-Control": "no-cache", + "Pragma": "no-cache", + "Expires": "Fri, 01 Jan 1980 00:00:00 GMT", + } + + +def _ensure_receive_auth(token: str | None) -> None: + ensure_active_user_if_auth_enabled(token) + + +@http_router.get("/{project:path}/info/refs") +def smart_http_info_refs( + project: ValidatedHttpProject, + service: Annotated[SmartHttpService, Query(description="Git Smart HTTP service name")], + token: Annotated[str | None, Depends(access_token_from_request)], +) -> Response: + if service == "git-receive-pack": + _ensure_receive_auth(token) + body = _service_advertisement(service, project) + advertise_type, _ = _SERVICE_CONTENT_TYPES[service] + return Response( + content=body, + media_type=advertise_type, + headers=_smart_http_cache_headers(), + ) + + +@http_router.post("/{project:path}/git-upload-pack") +async def smart_http_upload_pack( + project: ValidatedHttpProject, + request: Request, + content_encoding: Annotated[str | None, Header(alias="Content-Encoding")] = None, + git_protocol: Annotated[str | None, Header(alias="Git-Protocol")] = None, +) -> Response: + raw_body = await request.body() + body = _decode_request_body(raw_body, content_encoding) + result = _run_git_service( + "git-upload-pack", + project, + advertise=False, + body=body, + git_protocol=git_protocol, + ) + _, result_type = _SERVICE_CONTENT_TYPES["git-upload-pack"] + return Response(content=result, media_type=result_type, headers=_smart_http_cache_headers()) + + +@http_router.post("/{project:path}/git-receive-pack") +async def smart_http_receive_pack( + project: ValidatedHttpProject, + request: Request, + token: Annotated[str | None, Depends(access_token_from_request)], + content_encoding: Annotated[str | None, Header(alias="Content-Encoding")] = None, + git_protocol: Annotated[str | None, Header(alias="Git-Protocol")] = None, +) -> Response: + _ensure_receive_auth(token) + raw_body = await request.body() + body = _decode_request_body(raw_body, content_encoding) + result = _run_git_service( + "git-receive-pack", + project, + advertise=False, + body=body, + git_protocol=git_protocol, + ) + _, result_type = _SERVICE_CONTENT_TYPES["git-receive-pack"] + return Response(content=result, media_type=result_type, headers=_smart_http_cache_headers()) diff --git a/pygitweb/smart_http_test.py b/pygitweb/smart_http_test.py new file mode 100644 index 0000000..2195db9 --- /dev/null +++ b/pygitweb/smart_http_test.py @@ -0,0 +1,224 @@ +from __future__ import annotations + +import gzip +import socket +import subprocess +import threading +import time +from collections.abc import Generator +from pathlib import Path +from unittest.mock import patch + +import pygit2 +import pytest +import uvicorn +from fastapi.testclient import TestClient + +from pygitweb.auth_config import AuthConfig, LocalUser +from pygitweb.config import settings +from pygitweb.conftest import set_client_session +from pygitweb.main import app +from pygitweb.smart_http import _decode_request_body, _pkt_flush, _pkt_line, _service_advertisement + +_LOCAL_ADMIN = AuthConfig( + auth_mode="local", + local_users=[LocalUser(user="alice", password="secret")], +) + + +@pytest.fixture +def local_auth_config() -> Generator[AuthConfig, None, None]: + with ( + patch("pygitweb.auth_config.auth_config", _LOCAL_ADMIN), + patch("pygitweb.auth.auth_config", _LOCAL_ADMIN), + patch("pygitweb.main.auth_config", _LOCAL_ADMIN), + ): + yield _LOCAL_ADMIN + + +def _run_git(repo_dir: Path, *args: str) -> str: + completed = subprocess.run( + [settings.GIT, *args], + cwd=repo_dir, + capture_output=True, + text=True, + check=True, + ) + return completed.stdout.strip() + + +def _seed_bare_repo(root: Path, bare_dir: Path, message: str = "initial") -> None: + work_dir = root / "seed-work" + work_dir.mkdir() + _run_git(work_dir, "init", "-b", "main") + _run_git(work_dir, "config", "user.name", "tester") + _run_git(work_dir, "config", "user.email", "tester@example.com") + (work_dir / "README.md").write_text(f"{message}\n", encoding="utf-8") + _run_git(work_dir, "add", "README.md") + _run_git(work_dir, "commit", "-m", message) + _run_git(work_dir, "remote", "add", "origin", str(bare_dir)) + _run_git(work_dir, "push", "-u", "origin", "main") + + +@pytest.fixture(scope="class") +def http_repo_env(tmp_path_factory: pytest.TempPathFactory) -> Generator[dict[str, str], None, None]: + root = tmp_path_factory.mktemp("smart-http") + repo_dir = root / "demo" + pygit2.init_repository(str(repo_dir), bare=True) + _seed_bare_repo(root, repo_dir) + with ( + patch.object(settings, "PROJECTROOT", str(root)), + patch.object(settings, "PROJECTS_LIST", str(root)), + patch.object(settings, "PROJECT_MAXDEPTH", 2), + patch.object(settings, "STRICT_EXPORT", False), + patch.object(settings, "EXPORT_OK", ""), + patch.object(settings, "LIST_ALL", True), + patch.object(settings, "AUTH", False), + ): + yield {"project": "demo", "repo_dir": str(repo_dir), "root": str(root)} + + +class TestSmartHttpHelpers: + def test_pkt_line_and_flush(self) -> None: + assert _pkt_line(b"hello\n") == b"000ahello\n" + assert _pkt_flush() == b"0000" + + def test_decode_request_body(self) -> None: + raw = b"0014command=ls-refs\n0000" + assert _decode_request_body(raw, None) == raw + assert _decode_request_body(raw, "identity") == raw + assert _decode_request_body(gzip.compress(raw), "gzip") == raw + + def test_decode_request_body_rejects_invalid_gzip(self) -> None: + with pytest.raises(Exception) as exc_info: + _decode_request_body(b"not-gzip", "gzip") + assert exc_info.value.status_code == 400 + + def test_decode_request_body_rejects_unknown_encoding(self) -> None: + with pytest.raises(Exception) as exc_info: + _decode_request_body(b"payload", "br") + assert exc_info.value.status_code == 415 + + def test_service_advertisement_matches_git(self, http_repo_env: dict[str, str]) -> None: + direct = subprocess.run( + [settings.GIT, "upload-pack", "--advertise-refs", "."], + cwd=http_repo_env["repo_dir"], + capture_output=True, + check=True, + ).stdout + body = _service_advertisement("git-upload-pack", http_repo_env["project"]) + assert body.startswith(_pkt_line(b"# service=git-upload-pack\n")) + assert body.endswith(direct) + assert _pkt_flush() in body + + +class TestSmartHttpRoutes: + def test_upload_pack_info_refs(self, http_repo_env: dict[str, str]) -> None: + client = TestClient(app) + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-upload-pack"}, + ) + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("application/x-git-upload-pack-advertisement") + assert b"# service=git-upload-pack" in resp.content + assert b"refs/heads/" in resp.content + + def test_accepts_git_suffix(self, http_repo_env: dict[str, str]) -> None: + client = TestClient(app) + resp = client.get( + f"/http/{http_repo_env['project']}.git/info/refs", + params={"service": "git-upload-pack"}, + ) + assert resp.status_code == 200 + + def test_unknown_project_404(self, http_repo_env: dict[str, str]) -> None: + client = TestClient(app) + resp = client.get( + "/http/missing/info/refs", + params={"service": "git-upload-pack"}, + ) + assert resp.status_code == 404 + + def test_upload_pack_accepts_gzip_body(self, http_repo_env: dict[str, str]) -> None: + client = TestClient(app) + request_body = b"0000" + headers = { + "Content-Type": "application/x-git-upload-pack-request", + "Git-Protocol": "version=2", + } + plain = client.post( + f"/http/{http_repo_env['project']}/git-upload-pack", + content=request_body, + headers=headers, + ) + gzipped = client.post( + f"/http/{http_repo_env['project']}/git-upload-pack", + content=gzip.compress(request_body), + headers={**headers, "Content-Encoding": "gzip"}, + ) + assert plain.status_code == gzipped.status_code + assert plain.content == gzipped.content + + def test_receive_pack_requires_auth_when_enabled( + self, + http_repo_env: dict[str, str], + local_auth_config: AuthConfig, + ) -> None: + del local_auth_config + client = TestClient(app) + with patch.object(settings, "AUTH", True): + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-receive-pack"}, + ) + assert resp.status_code == 401 + + def test_receive_pack_allowed_when_authenticated( + self, + http_repo_env: dict[str, str], + local_auth_config: AuthConfig, + ) -> None: + del local_auth_config + client = TestClient(app) + with patch.object(settings, "AUTH", True): + set_client_session(client, "alice") + resp = client.get( + f"/http/{http_repo_env['project']}/info/refs", + params={"service": "git-receive-pack"}, + ) + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("application/x-git-receive-pack-advertisement") + + def test_clone_and_push_over_http(self, http_repo_env: dict[str, str]) -> None: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + + config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error") + server = uvicorn.Server(config) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + deadline = time.time() + 5.0 + while not server.started and time.time() < deadline: + time.sleep(0.01) + assert server.started + + clone_dir = Path(http_repo_env["root"]) / "clone" + remote_url = f"http://127.0.0.1:{port}/http/{http_repo_env['project']}.git" + try: + _run_git(Path(http_repo_env["root"]), "clone", remote_url, str(clone_dir)) + _run_git(clone_dir, "config", "user.name", "tester") + _run_git(clone_dir, "config", "user.email", "tester@example.com") + (clone_dir / "new.txt").write_text("pushed\n", encoding="utf-8") + _run_git(clone_dir, "add", "new.txt") + _run_git(clone_dir, "commit", "-m", "add new") + _run_git(clone_dir, "push", "origin", "HEAD") + repo = pygit2.Repository(http_repo_env["repo_dir"]) + commit = repo.head.peel() + assert commit.message.strip() == "add new" + tree = commit.tree + assert tree["new.txt"].data.decode("utf-8") == "pushed\n" + finally: + server.should_exit = True + thread.join(timeout=5.0)