From 206a8730a892076f5b5395bd49089a15af90835e Mon Sep 17 00:00:00 2001 From: Aleksey Lobanov Date: Sat, 27 Jun 2026 22:57:58 +0300 Subject: [PATCH] feat!: token support for cloning --- Readme.md | 13 ++-- config.example.toml | 8 ++- gitea-mirror.py | 18 ++++-- src/config.py | 18 +++--- src/git.py | 57 ++++++++++++----- src/gitea_api.py | 8 +-- src/models.py | 24 ++++++- src/repository_name.py | 1 + tests/test_config.py | 43 +++++++++++-- tests/test_git.py | 72 +++++++++++++++++++++ tests/test_mirror_process.py | 119 +++++++++++++++++++++++++++++++++++ 11 files changed, 336 insertions(+), 45 deletions(-) create mode 100644 tests/test_git.py create mode 100644 tests/test_mirror_process.py diff --git a/Readme.md b/Readme.md index 3b18305..354f3ce 100644 --- a/Readme.md +++ b/Readme.md @@ -20,11 +20,12 @@ Other methods are not supporting: Which is acceptable for full account mirroring. **Security notice.** -This application uses SSH as git transport layer. -It is safe enough with right use, -and for right use you need to save -git server ssh digest (~/.ssh/known_hosts file). -To do this you just need to clone any repository over ssh first +This application can use SSH or HTTP(S) as git transport layer. +With SSH, you need to save git server ssh digest +(~/.ssh/known_hosts file). To do this you just need to clone any +repository over ssh first. +With HTTP(S), credentials are passed to git through environment-based +configuration and are not written into repository remote URLs. **Config**. We use single config for this application. It is slightly ancient solution for modern Docker/Kubernetes backends, @@ -32,6 +33,8 @@ but provides configuration in one place and _secure enough_ place to save token. Example config is available in [config.example.toml](config.example.toml). +Use `main.api_token` for Gitea API access. Use `fetch.ssh_key` for SSH +cloning, or `fetch.user` and `fetch.token` for HTTP(S) cloning. ### Native diff --git a/config.example.toml b/config.example.toml index 607b7ef..ea98235 100644 --- a/config.example.toml +++ b/config.example.toml @@ -1,6 +1,12 @@ [main] endpoint = "https://example.com" -token = "XXXXX" +api_token = "XXXXX" format = "{owner}/{name}" out_dir = "/home/user/repositories" + +[fetch] ssh_key = "/home/user/id_rsa" + +# Use this fetch config instead of ssh_key to clone over HTTP(S): +# user = "gitea-user" +# token = "XXXXX" diff --git a/gitea-mirror.py b/gitea-mirror.py index 62f2788..18a0f5b 100644 --- a/gitea-mirror.py +++ b/gitea-mirror.py @@ -3,7 +3,7 @@ import sys from os import makedirs from src.config import Config, read_toml_config -from src.git import git_clone, git_pull +from src.git import git_clone, git_pull, http_git_env, ssh_git_env from src.gitea_api import GiteaApi from src.models import GiteaRepository from src.repository_name import get_repository_name, is_valid_repository_names @@ -12,15 +12,25 @@ BASE_PATH = "out" FORMAT = "{owner}/{name}" +def get_fetch_params(config: Config, repo: GiteaRepository): + if config.fetch.ssh_key_path: + return repo.ssh_url, ssh_git_env(config.fetch.ssh_key_path) + return repo.clone_url, http_git_env( + user=config.fetch.user, + token=config.fetch.token, + ) + + def process_repo(config: Config, repo: GiteaRepository): path = get_repository_name(name_format=config.repository_format, r=repo) out_path = os.path.join(config.out_dir, path) + repository_url, git_env = get_fetch_params(config=config, repo=repo) makedirs(out_path, exist_ok=True) if os.path.exists(os.path.join(out_path, ".git")): - git_pull(out_path, ssh_key="fake") + git_pull(out_path, env=git_env) return print(f"New repository: {path}") - git_clone(ssh_url=repo.ssh_url, repository=out_path, ssh_key="fake") + git_clone(repository_url=repository_url, repository=out_path, env=git_env) def main(): @@ -35,7 +45,7 @@ def main(): api = GiteaApi( endpoint=config.endpoint, - token=config.token, + api_token=config.api_token, ) repos = api.get_repositories() print(f"total {len(repos)} repositories") diff --git a/src/config.py b/src/config.py index 09947c4..12bf252 100644 --- a/src/config.py +++ b/src/config.py @@ -5,17 +5,19 @@ files are more secure in general than command-line arguments TOML config example [main] endpoint = "https://example.com/gitea" -token = "something" +api_token = "something" format = "{owner}/{name}" out_dir = "/home/user/repositories" -ssh_key = "/home/user/.ssh/id_rsa.pub" + +[fetch] +ssh_key = "/home/user/.ssh/id_rsa" """ import os import tomllib -from .models import Config +from .models import Config, FetchConfig MAIN_SECTION = "main" @@ -29,17 +31,17 @@ def read_toml_config(path: str) -> Config: data = tomllib.load(config_file) main = data[MAIN_SECTION] endpoint = main["endpoint"] - token = main["token"] + api_token = main["api_token"] repository_format = main["format"] out_dir = main["out_dir"] - ssh_key_path = main["ssh_key"] - except (KeyError, tomllib.TOMLDecodeError) as err: + fetch = FetchConfig(**data["fetch"]) + except (KeyError, ValueError, tomllib.TOMLDecodeError) as err: raise RuntimeError(f"Invalid TOML config: {err}") from err return Config( repository_format=repository_format, endpoint=endpoint, - token=token, + api_token=api_token, out_dir=out_dir, - ssh_key_path=ssh_key_path, + fetch=fetch, ) diff --git a/src/git.py b/src/git.py index 5c99e85..1e14a08 100644 --- a/src/git.py +++ b/src/git.py @@ -1,24 +1,53 @@ +import base64 +import os +import shlex import subprocess from os import makedirs -def git_clone(ssh_url: str, repository: str, ssh_key: str) -> bool: +def ssh_git_env(ssh_key_path: str) -> dict[str, str]: + return { + "GIT_SSH_COMMAND": ( + f"ssh -i {shlex.quote(ssh_key_path)} -o IdentitiesOnly=yes" + ), + } + + +def http_git_env(user: str, token: str) -> dict[str, str]: + credentials = base64.b64encode(f"{user}:{token}".encode("utf-8")).decode( + "ascii" + ) + return { + "GIT_CONFIG_COUNT": "1", + "GIT_CONFIG_KEY_0": "http.extraHeader", + "GIT_CONFIG_VALUE_0": f"Authorization: Basic {credentials}", + } + + +def git_clone( + repository_url: str, + repository: str, + env: dict[str, str] | None = None, +) -> bool: makedirs(repository, exist_ok=True) + git_env = os.environ | (env or {}) try: - subprocess.check_call(["git", "clone", ssh_url, "."], cwd=repository) - except subprocess.CalledProcessError: - print( - "Unable to clone repository " - f"{repository} with key {ssh_key} from {ssh_url}" + subprocess.check_call( + ["git", "clone", repository_url, "."], + cwd=repository, + env=git_env, ) - return False - return True - - -def git_pull(repository: str, ssh_key: str) -> bool: - try: - subprocess.check_call(["git", "pull"], cwd=repository) except subprocess.CalledProcessError: - print(f"Unable to pull repository {repository} with key {ssh_key}") + print(f"Unable to clone repository {repository} from {repository_url}") + return False + return True + + +def git_pull(repository: str, env: dict[str, str] | None = None) -> bool: + git_env = os.environ | (env or {}) + try: + subprocess.check_call(["git", "pull"], cwd=repository, env=git_env) + except subprocess.CalledProcessError: + print(f"Unable to pull repository {repository}") return False return True diff --git a/src/gitea_api.py b/src/gitea_api.py index 49d7892..19c9838 100644 --- a/src/gitea_api.py +++ b/src/gitea_api.py @@ -9,16 +9,16 @@ REPOSITORIES_ADAPTER = TypeAdapter(list[GiteaRepository]) class GiteaApi: - def __init__(self, endpoint: str, token: str): - self._endpoint = endpoint - self._token = token + def __init__(self, endpoint: str, api_token: str): + self._endpoint = str(endpoint) + self._api_token = api_token def get_repositories(self, page_size=10) -> list[GiteaRepository]: """ For mirroring input user is not important. """ session = requests.session() - session.headers.update({"Authorization": "token " + self._token}) + session.headers.update({"Authorization": "token " + self._api_token}) all_repos = {} # hack for unique repositories in result page_id = 1 while True: diff --git a/src/models.py b/src/models.py index a4b8c46..fb46ff4 100644 --- a/src/models.py +++ b/src/models.py @@ -1,14 +1,31 @@ import datetime -from pydantic import BaseModel, Field, HttpUrl +from pydantic import BaseModel, Field, HttpUrl, model_validator + + +class FetchConfig(BaseModel): + ssh_key_path: str | None = Field(default=None, alias="ssh_key") + user: str | None = None + token: str | None = None + + @model_validator(mode="after") + def validate_fetch_method(self): + has_ssh = self.ssh_key_path is not None + has_http = self.user is not None or self.token is not None + + if has_ssh and has_http: + raise ValueError("fetch must use either ssh_key or user/token") + if not has_ssh and not (self.user and self.token): + raise ValueError("fetch must define ssh_key or user/token") + return self class Config(BaseModel): repository_format: str - ssh_key_path: str endpoint: HttpUrl - token: str + api_token: str out_dir: str + fetch: FetchConfig class GiteaUser(BaseModel): @@ -19,6 +36,7 @@ class GiteaUser(BaseModel): class GiteaRepository(BaseModel): ssh_url: str + clone_url: str name: str repo_id: int = Field(alias="id") updated_at: datetime.datetime diff --git a/src/repository_name.py b/src/repository_name.py index 2fac7ae..d9c46b2 100644 --- a/src/repository_name.py +++ b/src/repository_name.py @@ -7,6 +7,7 @@ from .models import GiteaRepository, GiteaUser def _get_test_repository() -> GiteaRepository: return GiteaRepository( ssh_url="ssh://git@example.com/project/name", + clone_url="https://example.com/project/name.git", name="test name", id=42, updated_at=datetime.datetime.now(), diff --git a/tests/test_config.py b/tests/test_config.py index 12c58fa..4830f15 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -2,27 +2,58 @@ from tempfile import NamedTemporaryFile import pytest -from src.config import Config, read_toml_config +from src.config import Config, FetchConfig, read_toml_config @pytest.mark.parametrize( "config_data, expected", [ ( - '[main]\ntoken = "something"\n' + '[main]\napi_token = "something"\n' 'format = "{owner}/{name}"\n' - 'ssh_key = "/tmp/no_key"\n' 'endpoint = "https://example.com"\n' - 'out_dir = "/home/user/repositories"', + 'out_dir = "/home/user/repositories"\n' + "\n" + "[fetch]\n" + 'ssh_key = "/tmp/no_key"', Config( - token="something", + api_token="something", repository_format="{owner}/{name}", out_dir="/home/user/repositories", endpoint="https://example.com", - ssh_key_path="/tmp/no_key", + fetch=FetchConfig(ssh_key="/tmp/no_key"), + ), + ), + ( + '[main]\napi_token = "something"\n' + 'format = "{owner}/{name}"\n' + 'endpoint = "https://example.com"\n' + 'out_dir = "/home/user/repositories"\n' + "\n" + "[fetch]\n" + 'user = "alex"\n' + 'token = "repo-token"', + Config( + api_token="something", + repository_format="{owner}/{name}", + out_dir="/home/user/repositories", + endpoint="https://example.com", + fetch=FetchConfig(user="alex", token="repo-token"), ), ), ("[main]", None), + ( + '[main]\napi_token = "something"\n' + 'format = "{owner}/{name}"\n' + 'endpoint = "https://example.com"\n' + 'out_dir = "/home/user/repositories"\n' + "\n" + "[fetch]\n" + 'ssh_key = "/tmp/no_key"\n' + 'user = "alex"\n' + 'token = "repo-token"', + None, + ), ("not valid toml", None), ], ) diff --git a/tests/test_git.py b/tests/test_git.py new file mode 100644 index 0000000..a414aaf --- /dev/null +++ b/tests/test_git.py @@ -0,0 +1,72 @@ +import base64 +import os +import subprocess + +from src.git import git_clone, git_pull, http_git_env, ssh_git_env + + +def test_ssh_git_env_uses_requested_identity_file(): + env = ssh_git_env("/tmp/test_key") + + assert env == { + "GIT_SSH_COMMAND": "ssh -i /tmp/test_key -o IdentitiesOnly=yes" + } + + +def test_ssh_git_env_quotes_identity_file_path(): + env = ssh_git_env("/tmp/test key") + + assert env == { + "GIT_SSH_COMMAND": "ssh -i '/tmp/test key' -o IdentitiesOnly=yes" + } + + +def test_http_git_env_uses_basic_auth_header_without_url_credentials(): + env = http_git_env(user="alex", token="secret") + expected_credentials = base64.b64encode(b"alex:secret").decode("ascii") + + assert env == { + "GIT_CONFIG_COUNT": "1", + "GIT_CONFIG_KEY_0": "http.extraHeader", + "GIT_CONFIG_VALUE_0": f"Authorization: Basic {expected_credentials}", + } + + +def test_git_clone_passes_url_and_env_to_git(monkeypatch, tmp_path): + calls = [] + env = {"GIT_SSH_COMMAND": "ssh -i /tmp/test_key -o IdentitiesOnly=yes"} + + def fake_check_call(args, cwd, env): + calls.append((args, cwd, env)) + + monkeypatch.setattr(subprocess, "check_call", fake_check_call) + + assert git_clone( + repository_url="ssh://git@example.com/owner/repo.git", + repository=str(tmp_path), + env=env, + ) + + args, cwd, actual_env = calls[0] + assert args == ["git", "clone", "ssh://git@example.com/owner/repo.git", "."] + assert cwd == str(tmp_path) + assert os.environ.items() <= actual_env.items() + assert env.items() <= actual_env.items() + + +def test_git_pull_passes_env_to_git(monkeypatch, tmp_path): + calls = [] + env = {"GIT_CONFIG_COUNT": "1"} + + def fake_check_call(args, cwd, env): + calls.append((args, cwd, env)) + + monkeypatch.setattr(subprocess, "check_call", fake_check_call) + + assert git_pull(repository=str(tmp_path), env=env) + + args, cwd, actual_env = calls[0] + assert args == ["git", "pull"] + assert cwd == str(tmp_path) + assert os.environ.items() <= actual_env.items() + assert env.items() <= actual_env.items() diff --git a/tests/test_mirror_process.py b/tests/test_mirror_process.py new file mode 100644 index 0000000..88b3586 --- /dev/null +++ b/tests/test_mirror_process.py @@ -0,0 +1,119 @@ +import datetime +import importlib.util +from pathlib import Path + +from src.config import Config, FetchConfig +from src.models import GiteaRepository, GiteaUser + + +def load_mirror_module(): + module_path = Path(__file__).parent.parent / "gitea-mirror.py" + spec = importlib.util.spec_from_file_location("gitea_mirror", module_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def make_repo() -> GiteaRepository: + return GiteaRepository( + ssh_url="ssh://git@example.com/alex/repo.git", + clone_url="https://example.com/alex/repo.git", + name="repo", + id=42, + updated_at=datetime.datetime.now(), + owner=GiteaUser( + id=23, + login="alex", + email="alex@example.com", + ), + ) + + +def make_config(tmp_path, fetch: FetchConfig) -> Config: + return Config( + api_token="api-token", + endpoint="https://example.com", + repository_format="{owner}/{name}", + out_dir=str(tmp_path), + fetch=fetch, + ) + + +def test_process_repo_clones_with_ssh_url_and_key_env(monkeypatch, tmp_path): + mirror = load_mirror_module() + calls = [] + + def fake_git_clone(repository_url, repository, env): + calls.append((repository_url, repository, env)) + return True + + monkeypatch.setattr(mirror, "git_clone", fake_git_clone) + + mirror.process_repo( + config=make_config(tmp_path, FetchConfig(ssh_key="/tmp/test_key")), + repo=make_repo(), + ) + + assert calls == [ + ( + "ssh://git@example.com/alex/repo.git", + str(tmp_path / "alex" / "repo"), + { + "GIT_SSH_COMMAND": ( + "ssh -i /tmp/test_key -o IdentitiesOnly=yes" + ), + }, + ) + ] + + +def test_process_repo_clones_with_https_url_and_auth_env(monkeypatch, tmp_path): + mirror = load_mirror_module() + calls = [] + + def fake_git_clone(repository_url, repository, env): + calls.append((repository_url, repository, env)) + return True + + monkeypatch.setattr(mirror, "git_clone", fake_git_clone) + + mirror.process_repo( + config=make_config(tmp_path, FetchConfig(user="alex", token="secret")), + repo=make_repo(), + ) + + repository_url, repository, env = calls[0] + assert repository_url == "https://example.com/alex/repo.git" + assert repository == str(tmp_path / "alex" / "repo") + assert env["GIT_CONFIG_COUNT"] == "1" + assert env["GIT_CONFIG_KEY_0"] == "http.extraHeader" + assert env["GIT_CONFIG_VALUE_0"].startswith("Authorization: Basic ") + + +def test_process_repo_pulls_existing_repo_with_fetch_env(monkeypatch, tmp_path): + mirror = load_mirror_module() + calls = [] + repo_path = tmp_path / "alex" / "repo" / ".git" + repo_path.mkdir(parents=True) + + def fake_git_pull(repository, env): + calls.append((repository, env)) + return True + + monkeypatch.setattr(mirror, "git_pull", fake_git_pull) + + mirror.process_repo( + config=make_config(tmp_path, FetchConfig(ssh_key="/tmp/test_key")), + repo=make_repo(), + ) + + assert calls == [ + ( + str(tmp_path / "alex" / "repo"), + { + "GIT_SSH_COMMAND": ( + "ssh -i /tmp/test_key -o IdentitiesOnly=yes" + ), + }, + ) + ]