You're looking at it

Homepage: https://roundabout-host.com

By using this site, you agree to have cookies stored on your device, strictly for functional purposes, such as storing your session and preferences.

Dismiss

 git_http.py

View raw Download
text/plain • 7.72 kiB
Python script, ASCII text executable
        
            
1
"""
2
This module provides a Flask implementation of the Git smart HTTP protocol.
3
4
Roundabout - git hosting for everyone <https://roundabout-host.com>
5
Copyright (C) 2023-2025 Roundabout developers <root@roundabout-host.com>
6
7
This program is free software: you can redistribute it and/or modify
8
it under the terms of the GNU Affero General Public License as published by
9
the Free Software Foundation, either version 3 of the License, or
10
(at your option) any later version.
11
12
This program is distributed in the hope that it will be useful,
13
but WITHOUT ANY WARRANTY; without even the implied warranty of
14
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15
GNU Affero General Public License for more details.
16
17
You should have received a copy of the GNU Affero General Public License
18
along with this program. If not, see <http://www.gnu.org/licenses/>.
19
"""
20
21
import uuid
22
from models import *
23
from app import app, db, bcrypt
24
from misc_utils import *
25
from common import git_command
26
import os
27
import shutil
28
import config
29
import flask
30
import git
31
import subprocess
32
from flask_httpauth import HTTPBasicAuth
33
import zlib
34
import re
35
import datetime
36
37
auth = HTTPBasicAuth(realm=config.AUTH_REALM)
38
39
auth_required = flask.Response("Unauthorized Access", 401,
40
{"WWW-Authenticate": 'Basic realm="Login Required"'})
41
42
43
@auth.verify_password
44
def verify_password(username, password):
45
user = User.query.filter_by(username=username).first()
46
47
if user and bcrypt.check_password_hash(user.password_hashed, password):
48
flask.g.user = username
49
return True
50
51
return False
52
53
54
@app.route("/<username>/<repository>/git-upload-pack", methods=["POST"])
55
@app.route("/git/<username>/<repository>/git-upload-pack", methods=["POST"])
56
@auth.login_required(optional=True)
57
def git_upload_pack(username, repository):
58
server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git")
59
if not os.path.exists(server_repo_location):
60
flask.abort(404)
61
62
if auth.current_user() is None and not get_visibility(username, repository):
63
return auth_required
64
if not (get_visibility(username, repository) or get_permission_level(flask.g.user, username,
65
repository) is not None):
66
flask.abort(403)
67
68
text = git_command(server_repo_location, flask.request.data, "upload-pack",
69
"--stateless-rpc", ".")
70
71
return flask.Response(text, content_type="application/x-git-upload-pack-result")
72
73
74
@app.route("/<username>/<repository>/git-receive-pack", methods=["POST"])
75
@app.route("/git/<username>/<repository>/git-receive-pack", methods=["POST"])
76
@auth.login_required
77
def git_receive_pack(username, repository):
78
server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git")
79
if not os.path.exists(server_repo_location):
80
flask.abort(404)
81
82
if not get_permission_level(flask.g.user, username, repository):
83
flask.abort(403)
84
85
server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git")
86
text = git_command(server_repo_location, flask.request.data, "receive-pack",
87
"--stateless-rpc", ".")
88
89
if flask.request.data == b"0000":
90
return flask.Response("", content_type="application/x-git-receive-pack-result")
91
92
push_info = flask.request.data.split(b"\x00")[0].decode()
93
if not push_info:
94
return flask.Response(text, content_type="application/x-git-receive-pack-result")
95
96
old_sha, new_sha, ref = push_info[4:].split() # discard first 4 characters, used for line length which we don't need
97
98
if old_sha == "0" * 40:
99
commits_list = subprocess.check_output(["git", "rev-list", new_sha],
100
cwd=server_repo_location).decode().strip().split("\n")
101
else:
102
commits_list = subprocess.check_output(["git", "rev-list", f"{old_sha}..{new_sha}"],
103
cwd=server_repo_location).decode().strip().split("\n")
104
105
for sha in reversed(commits_list):
106
info = git_command(server_repo_location, None, "show", "-s",
107
"--format='%H%n%at%n%cn <%ce>%n%B'", sha).decode()
108
109
sha, time, identity, body = info.split("\n", 3)
110
login = flask.g.user
111
112
if not Commit.query.filter_by(identifier=f"/{username}/{repository}/{sha}").first():
113
logged_in_user = User.query.filter_by(username=login).first()
114
user = get_commit_identity(identity, logged_in_user, db.session.get(Repo, f"/{username}/{repository}"))
115
repo = Repo.query.filter_by(route=f"/{username}/{repository}").first()
116
117
commit = Commit(sha, user, repo, time, body, identity, logged_in_user)
118
119
db.session.add(commit)
120
db.session.commit()
121
122
if ref.startswith("refs/heads/"): # if the push is to a branch
123
ref = ref.rpartition("/")[2] # get the branch name only
124
repo_data = db.session.get(Repo, f"/{username}/{repository}")
125
if ref == repo_data.site_branch:
126
# Update the site
127
from celery_tasks import copy_site
128
copy_site.delay(repo_data.route)
129
130
return flask.Response(text, content_type="application/x-git-receive-pack-result")
131
132
133
@app.route("/<username>/<repository>/info/refs", methods=["GET", "POST"])
134
@app.route("/git/<username>/<repository>/info/refs", methods=["GET", "POST"])
135
@auth.login_required(optional=True)
136
def git_info_refs(username, repository):
137
server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git")
138
if not os.path.exists(server_repo_location):
139
flask.abort(404)
140
141
repo = git.Repo(server_repo_location)
142
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
143
if not repo_data.default_branch:
144
if repo.heads:
145
repo_data.default_branch = repo.heads[0].name
146
repo.git.checkout("-f", repo_data.default_branch)
147
148
if auth.current_user() is None and (
149
not get_visibility(username, repository) or flask.request.args.get(
150
"service") == "git-receive-pack"):
151
return auth_required
152
try:
153
if not (get_visibility(username, repository) or get_permission_level(flask.g.user,
154
username,
155
repository) is not None):
156
flask.abort(403)
157
except AttributeError:
158
return auth_required
159
160
service = flask.request.args.get("service")
161
162
if service.startswith("git"):
163
service = service[4:]
164
else:
165
flask.abort(403)
166
167
if service == "receive-pack":
168
try:
169
if not get_permission_level(flask.g.user, username, repository):
170
flask.abort(403)
171
except AttributeError:
172
return auth_required
173
174
service_line = f"# service=git-{service}\n"
175
service_line = (f"{len(service_line) + 4:04x}" + service_line).encode()
176
177
if service == "upload-pack":
178
text = service_line + b"0000" + git_command(server_repo_location, None, "upload-pack",
179
"--stateless-rpc",
180
"--advertise-refs",
181
"--http-backend-info-refs", ".")
182
elif service == "receive-pack":
183
refs = git_command(server_repo_location, None, "receive-pack",
184
"--http-backend-info-refs", ".")
185
text = service_line + b"0000" + refs
186
else:
187
flask.abort(403)
188
189
response = flask.Response(text, content_type=f"application/x-git-{service}-advertisement")
190
response.headers["Cache-Control"] = "no-cache"
191
192
return response
193