alex/gitea-mirror
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

feat!: token support for cloning

Browse Source
This commit is contained in:
Aleksey Lobanov
2026-06-27 22:57:58 +03:00
parent 28f1fda048
commit 206a8730a8
11 changed files with 336 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Readme.md
+8 -5
View File
@@ -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
config.example.toml
+7 -1
View File
@@ -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"
gitea-mirror.py
+14 -4
View File
@@ -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")
src/config.py
+10 -8
View File
@@ -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,
)
src/git.py
+42 -13
View File
@@ -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:
makedirs(repository, exist_ok=True)
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}"
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 False
return True
return {
"GIT_CONFIG_COUNT": "1",
"GIT_CONFIG_KEY_0": "http.extraHeader",
"GIT_CONFIG_VALUE_0": f"Authorization: Basic {credentials}",
}
def git_pull(repository: str, ssh_key: str) -> bool:
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", "pull"], cwd=repository)
subprocess.check_call(
["git", "clone", repository_url, "."],
cwd=repository,
env=git_env,
)
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
src/gitea_api.py
+4 -4
View File
@@ -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:
src/models.py
+21 -3
View File
@@ -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
src/repository_name.py
+1
View File
@@ -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(),
tests/test_config.py
+37 -6
View File
@@ -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),
],
)
tests/test_git.py
+72
View File
@@ -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()
tests/test_mirror_process.py
+119
View File
@@ -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"
),
},
)
]