551 lines
24 KiB
Python
551 lines
24 KiB
Python
import asyncio
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
|
|
from aiohttp import web
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger("git_server")
|
|
|
|
|
|
class GitApplication(web.Application):
|
|
def __init__(self, parent=None):
|
|
# import git
|
|
# globals()['git'] = git
|
|
self.parent = parent
|
|
super().__init__(client_max_size=1024 * 1024 * 1024 * 5)
|
|
self.add_routes(
|
|
[
|
|
web.post("/create/{repo_name}", self.create_repository),
|
|
web.delete("/delete/{repo_name}", self.delete_repository),
|
|
web.get("/clone/{repo_name}", self.clone_repository),
|
|
web.post("/push/{repo_name}", self.push_repository),
|
|
web.post("/pull/{repo_name}", self.pull_repository),
|
|
web.get("/status/{repo_name}", self.status_repository),
|
|
# web.get('/list', self.list_repositories),
|
|
web.get("/branches/{repo_name}", self.list_branches),
|
|
web.post("/branches/{repo_name}", self.create_branch),
|
|
web.get("/log/{repo_name}", self.commit_log),
|
|
web.get("/file/{repo_name}/{file_path:.*}", self.file_content),
|
|
web.get("/{path:.+}/info/refs", self.git_smart_http),
|
|
web.post("/{path:.+}/git-upload-pack", self.git_smart_http),
|
|
web.post("/{path:.+}/git-receive-pack", self.git_smart_http),
|
|
web.get("/{repo_name}.git/info/refs", self.git_smart_http),
|
|
web.post("/{repo_name}.git/git-upload-pack", self.git_smart_http),
|
|
web.post("/{repo_name}.git/git-receive-pack", self.git_smart_http),
|
|
]
|
|
)
|
|
|
|
async def check_basic_auth(self, request):
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if not auth_header.startswith("Basic "):
|
|
return None, None
|
|
encoded_creds = auth_header.split("Basic ")[1]
|
|
decoded_creds = base64.b64decode(encoded_creds).decode()
|
|
username, password = decoded_creds.split(":", 1)
|
|
request["user"] = await self.parent.services.user.authenticate(
|
|
username=username, password=password
|
|
)
|
|
if not request["user"]:
|
|
return None, None
|
|
request["repository_path"] = (
|
|
await self.parent.services.user.get_repository_path(request["user"]["uid"])
|
|
)
|
|
|
|
return request["user"]["username"], request["repository_path"]
|
|
|
|
@staticmethod
|
|
def require_auth(handler):
|
|
async def wrapped(self, request, *args, **kwargs):
|
|
username, repository_path = await self.check_basic_auth(request)
|
|
if not username or not repository_path:
|
|
return web.Response(
|
|
status=401,
|
|
headers={"WWW-Authenticate": "Basic"},
|
|
text="Authentication required",
|
|
)
|
|
request["username"] = username
|
|
request["repository_path"] = repository_path
|
|
return await handler(self, request, *args, **kwargs)
|
|
|
|
return wrapped
|
|
|
|
def repo_path(self, repository_path, repo_name):
|
|
return repository_path.joinpath(repo_name + ".git")
|
|
|
|
def check_repo_exists(self, repository_path, repo_name):
|
|
repo_dir = self.repo_path(repository_path, repo_name)
|
|
if not os.path.exists(repo_dir):
|
|
return web.Response(text="Repository not found", status=404)
|
|
return None
|
|
|
|
@require_auth
|
|
async def create_repository(self, request):
|
|
username = request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
repository_path = request["repository_path"]
|
|
if not repo_name or "/" in repo_name or ".." in repo_name:
|
|
return web.Response(text="Invalid repository name", status=400)
|
|
repo_dir = self.repo_path(repository_path, repo_name)
|
|
if os.path.exists(repo_dir):
|
|
return web.Response(text="Repository already exists", status=400)
|
|
try:
|
|
git.Repo.init(repo_dir, bare=True)
|
|
logger.info(f"Created repository: {repo_name} for user {username}")
|
|
return web.Response(text=f"Created repository {repo_name}")
|
|
except Exception as e:
|
|
logger.error(f"Error creating repository {repo_name}: {str(e)}")
|
|
return web.Response(text=f"Error creating repository: {str(e)}", status=500)
|
|
|
|
@require_auth
|
|
async def delete_repository(self, request):
|
|
username = request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
repository_path = request["repository_path"]
|
|
error_response = self.check_repo_exists(repository_path, repo_name)
|
|
if error_response:
|
|
return error_response
|
|
#'''
|
|
try:
|
|
shutil.rmtree(self.repo_path(repository_path, repo_name))
|
|
logger.info(f"Deleted repository: {repo_name} for user {username}")
|
|
return web.Response(text=f"Deleted repository {repo_name}")
|
|
except Exception as e:
|
|
logger.error(f"Error deleting repository {repo_name}: {str(e)}")
|
|
return web.Response(text=f"Error deleting repository: {str(e)}", status=500)
|
|
|
|
@require_auth
|
|
async def clone_repository(self, request):
|
|
request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
repository_path = request["repository_path"]
|
|
error_response = self.check_repo_exists(repository_path, repo_name)
|
|
if error_response:
|
|
return error_response
|
|
host = request.host
|
|
clone_url = f"http://{host}/{repo_name}.git"
|
|
response_data = {
|
|
"repository": repo_name,
|
|
"clone_command": f"git clone {clone_url}",
|
|
"clone_url": clone_url,
|
|
}
|
|
return web.json_response(response_data)
|
|
|
|
@require_auth
|
|
async def push_repository(self, request):
|
|
username = request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
repository_path = request["repository_path"]
|
|
error_response = self.check_repo_exists(repository_path, repo_name)
|
|
if error_response:
|
|
return error_response
|
|
try:
|
|
data = await request.json()
|
|
except json.JSONDecodeError:
|
|
return web.Response(text="Invalid JSON data", status=400)
|
|
commit_message = data.get("commit_message", "Update from server")
|
|
branch = data.get("branch", "main")
|
|
changes = data.get("changes", [])
|
|
if not changes:
|
|
return web.Response(text="No changes provided", status=400)
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
temp_repo = git.Repo.clone_from(
|
|
self.repo_path(repository_path, repo_name), temp_dir
|
|
)
|
|
for change in changes:
|
|
file_path = os.path.join(temp_dir, change.get("file", ""))
|
|
content = change.get("content", "")
|
|
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
|
with open(file_path, "w") as f:
|
|
f.write(content)
|
|
temp_repo.git.add(A=True)
|
|
if not temp_repo.config_reader().has_section("user"):
|
|
temp_repo.config_writer().set_value(
|
|
"user", "name", "Git Server"
|
|
).release()
|
|
temp_repo.config_writer().set_value(
|
|
"user", "email", "git@server.local"
|
|
).release()
|
|
temp_repo.index.commit(commit_message)
|
|
origin = temp_repo.remote("origin")
|
|
origin.push(refspec=f"{branch}:{branch}")
|
|
logger.info(f"Pushed to repository: {repo_name} for user {username}")
|
|
return web.Response(text=f"Successfully pushed changes to {repo_name}")
|
|
|
|
@require_auth
|
|
async def pull_repository(self, request):
|
|
username = request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
repository_path = request["repository_path"]
|
|
error_response = self.check_repo_exists(repository_path, repo_name)
|
|
if error_response:
|
|
return error_response
|
|
try:
|
|
data = await request.json()
|
|
except json.JSONDecodeError:
|
|
data = {}
|
|
remote_url = data.get("remote_url")
|
|
branch = data.get("branch", "main")
|
|
if not remote_url:
|
|
return web.Response(text="Remote URL is required", status=400)
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
try:
|
|
local_repo = git.Repo.clone_from(
|
|
self.repo_path(repository_path, repo_name), temp_dir
|
|
)
|
|
remote_name = "pull_source"
|
|
try:
|
|
remote = local_repo.create_remote(remote_name, remote_url)
|
|
except git.GitCommandError:
|
|
remote = local_repo.remote(remote_name)
|
|
remote.set_url(remote_url)
|
|
remote.fetch()
|
|
local_repo.git.merge(f"{remote_name}/{branch}")
|
|
origin = local_repo.remote("origin")
|
|
origin.push()
|
|
logger.info(
|
|
f"Pulled to repository {repo_name} from {remote_url} for user {username}"
|
|
)
|
|
return web.Response(
|
|
text=f"Successfully pulled changes from {remote_url} to {repo_name}"
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error pulling to {repo_name}: {str(e)}")
|
|
return web.Response(text=f"Error pulling changes: {str(e)}", status=500)
|
|
|
|
@require_auth
|
|
async def status_repository(self, request):
|
|
request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
repository_path = request["repository_path"]
|
|
error_response = self.check_repo_exists(repository_path, repo_name)
|
|
if error_response:
|
|
return error_response
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
try:
|
|
temp_repo = git.Repo.clone_from(
|
|
self.repo_path(repository_path, repo_name), temp_dir
|
|
)
|
|
branches = [b.name for b in temp_repo.branches]
|
|
active_branch = temp_repo.active_branch.name
|
|
commits = []
|
|
for commit in list(temp_repo.iter_commits(max_count=5)):
|
|
commits.append(
|
|
{
|
|
"id": commit.hexsha,
|
|
"author": f"{commit.author.name} <{commit.author.email}>",
|
|
"date": commit.committed_datetime.isoformat(),
|
|
"message": commit.message,
|
|
}
|
|
)
|
|
files = []
|
|
for root, dirs, filenames in os.walk(temp_dir):
|
|
if ".git" in root:
|
|
continue
|
|
for filename in filenames:
|
|
full_path = os.path.join(root, filename)
|
|
rel_path = os.path.relpath(full_path, temp_dir)
|
|
files.append(rel_path)
|
|
status_info = {
|
|
"repository": repo_name,
|
|
"branches": branches,
|
|
"active_branch": active_branch,
|
|
"recent_commits": commits,
|
|
"files": files,
|
|
}
|
|
return web.json_response(status_info)
|
|
except Exception as e:
|
|
logger.error(f"Error getting status for {repo_name}: {str(e)}")
|
|
return web.Response(
|
|
text=f"Error getting repository status: {str(e)}", status=500
|
|
)
|
|
|
|
@require_auth
|
|
async def list_repositories(self, request):
|
|
request["username"]
|
|
try:
|
|
repos = []
|
|
user_dir = self.REPO_DIR
|
|
if os.path.exists(user_dir):
|
|
for item in os.listdir(user_dir):
|
|
item_path = os.path.join(user_dir, item)
|
|
if os.path.isdir(item_path) and item.endswith(".git"):
|
|
repos.append(item[:-4])
|
|
if request.query.get("format") == "json":
|
|
return web.json_response({"repositories": repos})
|
|
else:
|
|
return web.Response(
|
|
text="\n".join(repos) if repos else "No repositories found"
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error listing repositories: {str(e)}")
|
|
return web.Response(
|
|
text=f"Error listing repositories: {str(e)}", status=500
|
|
)
|
|
|
|
@require_auth
|
|
async def list_branches(self, request):
|
|
request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
repository_path = request["repository_path"]
|
|
error_response = self.check_repo_exists(repository_path, repo_name)
|
|
if error_response:
|
|
return error_response
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
temp_repo = git.Repo.clone_from(
|
|
self.repo_path(repository_path, repo_name), temp_dir
|
|
)
|
|
branches = [b.name for b in temp_repo.branches]
|
|
return web.json_response({"branches": branches})
|
|
|
|
@require_auth
|
|
async def create_branch(self, request):
|
|
username = request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
repository_path = request["repository_path"]
|
|
error_response = self.check_repo_exists(repository_path, repo_name)
|
|
if error_response:
|
|
return error_response
|
|
try:
|
|
data = await request.json()
|
|
except json.JSONDecodeError:
|
|
return web.Response(text="Invalid JSON data", status=400)
|
|
branch_name = data.get("branch_name")
|
|
start_point = data.get("start_point", "HEAD")
|
|
if not branch_name:
|
|
return web.Response(text="Branch name is required", status=400)
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
try:
|
|
temp_repo = git.Repo.clone_from(
|
|
self.repo_path(repository_path, repo_name), temp_dir
|
|
)
|
|
temp_repo.git.branch(branch_name, start_point)
|
|
temp_repo.git.push("origin", branch_name)
|
|
logger.info(
|
|
f"Created branch {branch_name} in repository {repo_name} for user {username}"
|
|
)
|
|
return web.Response(text=f"Created branch {branch_name}")
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Error creating branch {branch_name} in {repo_name}: {str(e)}"
|
|
)
|
|
return web.Response(text=f"Error creating branch: {str(e)}", status=500)
|
|
|
|
@require_auth
|
|
async def commit_log(self, request):
|
|
request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
repository_path = request["repository_path"]
|
|
error_response = self.check_repo_exists(repository_path, repo_name)
|
|
if error_response:
|
|
return error_response
|
|
try:
|
|
limit = int(request.query.get("limit", 10))
|
|
branch = request.query.get("branch", "main")
|
|
except ValueError:
|
|
return web.Response(text="Invalid limit parameter", status=400)
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
try:
|
|
temp_repo = git.Repo.clone_from(
|
|
self.repo_path(repository_path, repo_name), temp_dir
|
|
)
|
|
commits = []
|
|
try:
|
|
for commit in list(temp_repo.iter_commits(branch, max_count=limit)):
|
|
commits.append(
|
|
{
|
|
"id": commit.hexsha,
|
|
"short_id": commit.hexsha[:7],
|
|
"author": f"{commit.author.name} <{commit.author.email}>",
|
|
"date": commit.committed_datetime.isoformat(),
|
|
"message": commit.message.strip(),
|
|
}
|
|
)
|
|
except git.GitCommandError as e:
|
|
if "unknown revision or path" in str(e):
|
|
commits = []
|
|
else:
|
|
raise
|
|
return web.json_response(
|
|
{"repository": repo_name, "branch": branch, "commits": commits}
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error getting commit log for {repo_name}: {str(e)}")
|
|
return web.Response(
|
|
text=f"Error getting commit log: {str(e)}", status=500
|
|
)
|
|
|
|
@require_auth
|
|
async def file_content(self, request):
|
|
request["username"]
|
|
repo_name = request.match_info["repo_name"]
|
|
file_path = request.match_info.get("file_path", "")
|
|
branch = request.query.get("branch", "main")
|
|
repository_path = request["repository_path"]
|
|
error_response = self.check_repo_exists(repository_path, repo_name)
|
|
if error_response:
|
|
return error_response
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
try:
|
|
temp_repo = git.Repo.clone_from(
|
|
self.repo_path(repository_path, repo_name), temp_dir
|
|
)
|
|
try:
|
|
temp_repo.git.checkout(branch)
|
|
except git.GitCommandError:
|
|
return web.Response(text=f"Branch '{branch}' not found", status=404)
|
|
file_full_path = os.path.join(temp_dir, file_path)
|
|
if not os.path.exists(file_full_path):
|
|
return web.Response(
|
|
text=f"File '{file_path}' not found", status=404
|
|
)
|
|
if os.path.isdir(file_full_path):
|
|
files = os.listdir(file_full_path)
|
|
return web.json_response(
|
|
{
|
|
"repository": repo_name,
|
|
"path": file_path,
|
|
"type": "directory",
|
|
"contents": files,
|
|
}
|
|
)
|
|
else:
|
|
try:
|
|
with open(file_full_path) as f:
|
|
content = f.read()
|
|
return web.Response(text=content)
|
|
except UnicodeDecodeError:
|
|
return web.Response(
|
|
text=f"Cannot display binary file content for '{file_path}'",
|
|
status=400,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error getting file content from {repo_name}: {str(e)}")
|
|
return web.Response(
|
|
text=f"Error getting file content: {str(e)}", status=500
|
|
)
|
|
|
|
@require_auth
|
|
async def git_smart_http(self, request):
|
|
request["username"]
|
|
repository_path = request["repository_path"]
|
|
path = request.path
|
|
|
|
async def get_repository_path():
|
|
req_path = path.lstrip("/")
|
|
if req_path.endswith("/info/refs"):
|
|
repo_name = req_path[: -len("/info/refs")]
|
|
elif req_path.endswith("/git-upload-pack"):
|
|
repo_name = req_path[: -len("/git-upload-pack")]
|
|
elif req_path.endswith("/git-receive-pack"):
|
|
repo_name = req_path[: -len("/git-receive-pack")]
|
|
else:
|
|
repo_name = req_path
|
|
if repo_name.endswith(".git"):
|
|
repo_name = repo_name[:-4]
|
|
repo_name = repo_name[4:]
|
|
repo_dir = repository_path.joinpath(repo_name + ".git")
|
|
logger.info(f"Resolved repo path: {repo_dir}")
|
|
return repo_dir
|
|
|
|
async def handle_info_refs(service):
|
|
repo_path = await get_repository_path()
|
|
|
|
logger.info(f"handle_info_refs: {repo_path}")
|
|
if not os.path.exists(repo_path):
|
|
return web.Response(text="Repository not found", status=404)
|
|
cmd = [service, "--stateless-rpc", "--advertise-refs", str(repo_path)]
|
|
try:
|
|
process = await asyncio.create_subprocess_exec(
|
|
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
|
|
)
|
|
stdout, stderr = await process.communicate()
|
|
if process.returncode != 0:
|
|
logger.error(f"Git command failed: {stderr.decode()}")
|
|
return web.Response(
|
|
text=f"Git error: {stderr.decode()}", status=500
|
|
)
|
|
response = web.StreamResponse(
|
|
status=200,
|
|
reason="OK",
|
|
headers={
|
|
"Content-Type": f"application/x-{service}-advertisement",
|
|
"Cache-Control": "no-cache",
|
|
},
|
|
)
|
|
await response.prepare(request)
|
|
packet = f"# service={service}\n"
|
|
length = len(packet) + 4
|
|
header = f"{length:04x}"
|
|
await response.write(f"{header}{packet}0000".encode())
|
|
await response.write(stdout)
|
|
return response
|
|
except Exception as e:
|
|
logger.error(f"Error handling info/refs: {str(e)}")
|
|
return web.Response(text=f"Server error: {str(e)}", status=500)
|
|
|
|
async def handle_service_rpc(service):
|
|
repo_path = await get_repository_path()
|
|
logger.info(f"handle_service_rpc: {repo_path}")
|
|
if not os.path.exists(repo_path):
|
|
return web.Response(text="Repository not found", status=404)
|
|
if (
|
|
not request.headers.get("Content-Type")
|
|
== f"application/x-{service}-request"
|
|
):
|
|
return web.Response(text="Invalid Content-Type", status=403)
|
|
body = await request.read()
|
|
cmd = [service, "--stateless-rpc", str(repo_path)]
|
|
try:
|
|
process = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdin=asyncio.subprocess.PIPE,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, stderr = await process.communicate(input=body)
|
|
if process.returncode != 0:
|
|
logger.error(f"Git command failed: {stderr.decode()}")
|
|
return web.Response(
|
|
text=f"Git error: {stderr.decode()}", status=500
|
|
)
|
|
return web.Response(
|
|
body=stdout, content_type=f"application/x-{service}-result"
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error handling service RPC: {str(e)}")
|
|
return web.Response(text=f"Server error: {str(e)}", status=500)
|
|
|
|
if request.method == "GET" and path.endswith("/info/refs"):
|
|
service = request.query.get("service")
|
|
if service in ("git-upload-pack", "git-receive-pack"):
|
|
return await handle_info_refs(service)
|
|
else:
|
|
return web.Response(
|
|
text="Smart HTTP requires service parameter", status=400
|
|
)
|
|
elif request.method == "POST" and "/git-upload-pack" in path:
|
|
return await handle_service_rpc("git-upload-pack")
|
|
elif request.method == "POST" and "/git-receive-pack" in path:
|
|
return await handle_service_rpc("git-receive-pack")
|
|
return web.Response(text="Not found", status=404)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
import uvloop
|
|
|
|
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
|
|
logger.info("Using uvloop for improved performance")
|
|
except ImportError:
|
|
logger.info("uvloop not available, using standard event loop")
|
|
app = GitApplication()
|
|
logger.info("Starting Git server on port 8080")
|
|
web.run_app(app, port=8080)
|