Python script, Unicode text, UTF-8 text executable
1
__version__ = "0.1.2"
2
import os
4
import shutil
5
import random
6
import subprocess
7
import platform
8
import git
9
import mimetypes
10
import magic
11
import flask
12
import cairosvg
13
import celery
14
import shlex
15
from functools import wraps
16
from datetime import datetime
17
from enum import Enum
18
from cairosvg import svg2png
19
from flask_sqlalchemy import SQLAlchemy
20
from flask_bcrypt import Bcrypt
21
from markupsafe import escape, Markup
22
from flask_migrate import Migrate
23
from PIL import Image
24
from flask_httpauth import HTTPBasicAuth
25
import config
26
from flask_babel import Babel, gettext, ngettext, force_locale
27
_ = gettext
29
n_ = gettext
30
app = flask.Flask(__name__)
32
app.config.from_mapping(
33
CELERY=dict(
34
broker_url=config.REDIS_URI,
35
result_backend=config.REDIS_URI,
36
task_ignore_result=True,
37
),
38
)
39
auth = HTTPBasicAuth()
41
app.config["SQLALCHEMY_DATABASE_URI"] = config.DB_URI
43
app.config["SECRET_KEY"] = config.DB_PASSWORD
44
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
45
app.config["BABEL_TRANSLATION_DIRECTORIES"] = "i18n"
46
app.config["MAX_CONTENT_LENGTH"] = config.MAX_PAYLOAD_SIZE
47
db = SQLAlchemy(app)
49
bcrypt = Bcrypt(app)
50
migrate = Migrate(app, db)
51
from models import *
53
from misc_utils import *
54
import git_http
56
import jinja_utils
57
import celery_tasks
58
from celery import Celery, Task
59
import celery_integration
60
import pathlib
61
babel = Babel(app)
63
def get_locale():
66
if flask.request.cookies.get("language"):
67
return flask.request.cookies.get("language")
68
return flask.request.accept_languages.best_match(config.available_locales)
69
babel.init_app(app, locale_selector=get_locale)
72
with app.app_context():
74
locale_names = {}
75
for language in config.available_locales:
76
with force_locale(language):
77
# NOTE: Translate this to the language's name in that language, for example in French you would use français
78
locale_names[language] = gettext("English")
79
worker = celery_integration.init_celery_app(app)
81
repositories = flask.Blueprint("repository", __name__, template_folder="templates/repository/")
83
app.jinja_env.add_extension("jinja2.ext.do")
85
app.jinja_env.add_extension("jinja2.ext.loopcontrols")
86
app.jinja_env.add_extension("jinja2.ext.debug")
87
@app.context_processor
90
def default():
91
username = flask.session.get("username")
92
user_object = User.query.filter_by(username=username).first()
94
return {
96
"logged_in_user": username,
97
"user_object": user_object,
98
"Notification": Notification,
99
"unread": UserNotification.query.filter_by(user_username=username).filter(
100
UserNotification.attention_level > 0).count(),
101
"config": config,
102
"Markup": Markup,
103
"locale_names": locale_names,
104
}
105
@app.route("/")
108
def main():
109
if flask.session.get("username"):
110
return flask.render_template("home.html")
111
else:
112
return flask.render_template("no-home.html")
113
@app.route("/userstyle")
116
def userstyle():
117
if flask.session.get("username") and os.path.exists(os.path.join(config.REPOS_PATH, flask.session.get("username"), ".config", "theme.css")):
118
return flask.send_from_directory(os.path.join(config.REPOS_PATH, flask.session.get("username"), ".config"), "theme.css")
119
else:
120
return flask.Response("", mimetype="text/css")
121
@app.route("/about/")
124
def about():
125
return flask.render_template("about.html", platform=platform, version=__version__)
126
@app.route("/language", methods=["POST"])
129
def set_locale():
130
response = flask.redirect(flask.request.referrer if flask.request.referrer else "/",
131
code=303)
132
if not flask.request.form.get("language"):
133
response.delete_cookie("language")
134
else:
135
response.set_cookie("language", flask.request.form.get("language"))
136
return response
138
@app.route("/cookie-dismiss")
141
def dismiss_banner():
142
response = flask.redirect(flask.request.referrer if flask.request.referrer else "/",
143
code=303)
144
response.set_cookie("cookie-banner", "1")
145
return response
146
@app.route("/help/")
149
def help_index():
150
return flask.render_template("help.html", faqs=config.faqs)
151
@app.route("/settings/", methods=["GET", "POST"])
154
def settings():
155
if not flask.session.get("username"):
156
flask.abort(401)
157
if flask.request.method == "GET":
158
user = User.query.filter_by(username=flask.session.get("username")).first()
159
return flask.render_template("user-settings.html", user=user)
161
else:
162
user = User.query.filter_by(username=flask.session.get("username")).first()
163
user.display_name = flask.request.form["displayname"]
165
user.URL = flask.request.form["url"]
166
user.company = flask.request.form["company"]
167
user.company_URL = flask.request.form["companyurl"]
168
user.email = flask.request.form.get("email") if flask.request.form.get(
169
"email") else None
170
user.location = flask.request.form["location"]
171
user.show_mail = True if flask.request.form.get("showmail") else False
172
user.bio = flask.request.form.get("bio")
173
db.session.commit()
175
flask.flash(
177
Markup("<iconify-icon icon='mdi:check'></iconify-icon>" + _("Settings saved")),
178
category="success")
179
return flask.redirect(f"/{flask.session.get('username')}", code=303)
180
@app.route("/favourites/", methods=["GET", "POST"])
183
def favourites():
184
if not flask.session.get("username"):
185
flask.abort(401)
186
if flask.request.method == "GET":
187
relationships = RepoFavourite.query.filter_by(
188
user_username=flask.session.get("username"))
189
return flask.render_template("favourites.html", favourites=relationships)
191
@app.route("/notifications/", methods=["GET", "POST"])
194
def notifications():
195
if not flask.session.get("username"):
196
flask.abort(401)
197
if flask.request.method == "GET":
198
return flask.render_template("notifications.html",
199
notifications=UserNotification.query.filter_by(
200
user_username=flask.session.get("username")))
201
@app.route("/notifications/<int:notification_id>/read", methods=["POST"])
204
def mark_read(notification_id):
205
if not flask.session.get("username"):
206
flask.abort(401)
207
notification = UserNotification.query.filter_by(id=notification_id).first()
208
if notification.user_username != flask.session.get("username"):
209
flask.abort(403)
210
notification.mark_read()
211
db.session.commit()
212
return f"<button hx-post='/notifications/{ notification.id }/unread' hx-swap='outerHTML'>Mark as unread</button>", 200
213
@app.route("/notifications/<int:notification_id>/unread", methods=["POST"])
216
def mark_unread(notification_id):
217
if not flask.session.get("username"):
218
flask.abort(401)
219
notification = UserNotification.query.filter_by(id=notification_id).first()
220
if notification.user_username != flask.session.get("username"):
221
flask.abort(403)
222
notification.mark_unread()
223
db.session.commit()
224
return f"<button hx-post='/notifications/{ notification.id }/read' hx-swap='outerHTML'>Mark as read</button>", 200
225
@app.route("/accounts/", methods=["GET", "POST"])
228
def login():
229
if flask.request.method == "GET":
230
return flask.render_template("login.html")
231
else:
232
if "login" in flask.request.form:
233
username = flask.request.form["username"]
234
password = flask.request.form["password"]
235
user = User.query.filter_by(username=username).first()
237
if user and bcrypt.check_password_hash(user.password_hashed, password):
239
flask.session["username"] = user.username
240
flask.flash(
241
Markup("<iconify-icon icon='mdi:account'></iconify-icon>" + _(
242
"Successfully logged in as {username}").format(username=username)),
243
category="success")
244
return flask.redirect("/", code=303)
245
elif not user:
246
flask.flash(Markup(
247
"<iconify-icon icon='mdi:account-question'></iconify-icon>" + _(
248
"User not found")),
249
category="alert")
250
return flask.render_template("login.html")
251
else:
252
flask.flash(Markup(
253
"<iconify-icon icon='mdi:account-question'></iconify-icon>" + _(
254
"Invalid password")),
255
category="error")
256
return flask.render_template("login.html")
257
if "signup" in flask.request.form:
258
username = flask.request.form["username"]
259
password = flask.request.form["password"]
260
password2 = flask.request.form["password2"]
261
email = flask.request.form.get("email")
262
email2 = flask.request.form.get("email2") # repeat email is a honeypot
263
name = flask.request.form.get("name")
264
if not only_chars(username,
266
"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"):
267
flask.flash(Markup(
268
_("Usernames may only contain Latin alphabet, numbers, '-' and '_'")),
269
category="error")
270
return flask.render_template("login.html")
271
if username in config.RESERVED_NAMES:
273
flask.flash(
274
Markup(
275
"<iconify-icon icon='mdi:account-error'></iconify-icon>" + _(
276
"Sorry, {username} is a system path").format(
277
username=username)),
278
category="error")
279
return flask.render_template("login.html")
280
user_check = User.query.filter_by(username=username).first()
282
if user_check or email2: # make the honeypot look like a normal error
283
flask.flash(
284
Markup(
285
"<iconify-icon icon='mdi:account-error'></iconify-icon>" + _(
286
"The username {username} is taken").format(
287
username=username)),
288
category="error")
289
return flask.render_template("login.html")
290
if password2 != password:
292
flask.flash(Markup("<iconify-icon icon='mdi:key-alert'></iconify-icon>" + _(
293
"Make sure the passwords match")),
294
category="error")
295
return flask.render_template("login.html")
296
user = User(username, password, email, name)
298
db.session.add(user)
299
db.session.commit()
300
flask.session["username"] = user.username
301
flask.flash(Markup(
302
"<iconify-icon icon='mdi:account'></iconify-icon>" + _(
303
"Successfully created and logged in as {username}").format(
304
username=username)),
305
category="success")
306
notification = Notification({"type": "welcome"})
308
db.session.add(notification)
309
db.session.commit()
310
result = celery_tasks.send_notification.delay(notification.id, [username], 1)
312
return flask.redirect("/", code=303)
314
@app.route("/newrepo/", methods=["GET", "POST"])
317
def new_repo():
318
if not flask.session.get("username"):
319
flask.abort(401)
320
if flask.request.method == "GET":
321
return flask.render_template("new-repo.html")
322
else:
323
name = flask.request.form["name"]
324
visibility = int(flask.request.form["visibility"])
325
if not only_chars(name,
327
"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"):
328
flask.flash(Markup(
329
"<iconify-icon icon='mdi:error'></iconify-icon>" + _(
330
"Repository names may only contain Latin alphabet, numbers, '-' and '_'")),
331
category="error")
332
return flask.render_template("new-repo.html")
333
user = User.query.filter_by(username=flask.session.get("username")).first()
335
repo = Repo(user, name, visibility)
337
db.session.add(repo)
338
db.session.commit()
339
flask.flash(Markup(_("Successfully created repository {name}").format(name=name)),
341
category="success")
342
return flask.redirect(repo.route, code=303)
343
@app.route("/logout")
346
def logout():
347
flask.session.clear()
348
flask.flash(Markup(
349
"<iconify-icon icon='mdi:account'></iconify-icon>" + _("Successfully logged out")),
350
category="info")
351
return flask.redirect("/", code=303)
352
@app.route("/<username>/", methods=["GET", "POST"])
355
def user_profile(username):
356
old_relationship = UserFollow.query.filter_by(
357
follower_username=flask.session.get("username"),
358
followed_username=username).first()
359
if flask.request.method == "GET":
360
user = User.query.filter_by(username=username).first()
361
match flask.request.args.get("action"):
362
case "repositories":
363
repos = Repo.query.filter_by(owner_name=username, visibility=2)
364
return flask.render_template("user-profile-repositories.html", user=user,
365
repos=repos,
366
relationship=old_relationship)
367
case "followers":
368
return flask.render_template("user-profile-followers.html", user=user,
369
relationship=old_relationship)
370
case "follows":
371
return flask.render_template("user-profile-follows.html", user=user,
372
relationship=old_relationship)
373
case _:
374
return flask.render_template("user-profile-overview.html", user=user,
375
relationship=old_relationship)
376
elif flask.request.method == "POST":
378
match flask.request.args.get("action"):
379
case "follow":
380
if username == flask.session.get("username"):
381
flask.abort(403)
382
if old_relationship:
383
db.session.delete(old_relationship)
384
else:
385
relationship = UserFollow(
386
flask.session.get("username"),
387
username
388
)
389
db.session.add(relationship)
390
db.session.commit()
391
user = db.session.get(User, username)
393
author = db.session.get(User, flask.session.get("username"))
394
notification = Notification({"type": "update", "version": "0.0.0"})
395
db.session.add(notification)
396
db.session.commit()
397
result = celery_tasks.send_notification.delay(notification.id, [username],
399
1)
400
db.session.commit()
402
return flask.redirect("?", code=303)
403
@app.route("/<username>/<repository>/")
406
def repository_index(username, repository):
407
return flask.redirect("./tree", code=302)
408
@app.route("/info/<username>/avatar")
411
def user_avatar(username):
412
serverUserdataLocation = os.path.join(config.USERDATA_PATH, username)
413
if not os.path.exists(serverUserdataLocation):
415
return flask.render_template("not-found.html"), 404
416
return flask.send_from_directory(serverUserdataLocation, "avatar.png")
418
@app.route("/<username>/<repository>/raw/<branch>/<path:subpath>")
421
def repository_raw(username, repository, branch, subpath):
422
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
423
if not os.path.exists(server_repo_location):
424
app.logger.error(f"Cannot load {server_repo_location}")
425
flask.abort(404)
426
if not (get_visibility(username, repository) or get_permission_level(
427
flask.session.get("username"), username,
428
repository) is not None):
429
flask.abort(403)
430
app.logger.info(f"Loading {server_repo_location}")
432
if not os.path.exists(server_repo_location):
434
app.logger.error(f"Cannot load {server_repo_location}")
435
return flask.render_template("not-found.html"), 404
436
repo = git.Repo(server_repo_location)
438
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
439
if not repo_data.default_branch:
440
if repo.heads:
441
repo_data.default_branch = repo.heads[0].name
442
else:
443
return flask.render_template("empty.html",
444
remote=f"http://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
445
if not branch:
446
branch = repo_data.default_branch
447
return flask.redirect(f"./{branch}", code=302)
448
if branch.startswith("tag:"):
450
ref = f"tags/{branch[4:]}"
451
elif branch.startswith("~"):
452
ref = branch[1:]
453
else:
454
ref = f"heads/{branch}"
455
ref = ref.replace("~", "/") # encode slashes for URL support
457
try:
459
repo.git.checkout("-f", ref)
460
except git.exc.GitCommandError:
461
return flask.render_template("not-found.html"), 404
462
return flask.send_from_directory(config.REPOS_PATH,
464
os.path.join(username, repository, subpath))
465
@repositories.route("/<username>/<repository>/tree/", defaults={"branch": None, "subpath": ""})
468
@repositories.route("/<username>/<repository>/tree/<branch>/", defaults={"subpath": ""})
469
@repositories.route("/<username>/<repository>/tree/<branch>/<path:subpath>")
470
def repository_tree(username, repository, branch, subpath):
471
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
472
if not os.path.exists(server_repo_location):
473
app.logger.error(f"Cannot load {server_repo_location}")
474
flask.abort(404)
475
if not (get_visibility(username, repository) or get_permission_level(
476
flask.session.get("username"), username,
477
repository) is not None):
478
flask.abort(403)
479
app.logger.info(f"Loading {server_repo_location}")
481
repo = git.Repo(server_repo_location)
483
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
484
if not repo_data.default_branch:
485
if repo.heads:
486
repo_data.default_branch = repo.heads[0].name
487
else:
488
return flask.render_template("empty.html",
489
remote=f"{config.www_protocol}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
490
if not branch:
491
branch = repo_data.default_branch
492
return flask.redirect(f"./{branch}", code=302)
493
if branch.startswith("tag:"):
495
ref = f"tags/{branch[4:]}"
496
elif branch.startswith("~"):
497
ref = branch[1:]
498
else:
499
ref = f"heads/{branch}"
500
ref = ref.replace("~", "/") # encode slashes for URL support
502
try:
504
repo.git.checkout("-f", ref)
505
except git.exc.GitCommandError:
506
return flask.render_template("not-found.html"), 404
507
branches = repo.heads
509
all_refs = []
511
for ref in repo.heads:
512
all_refs.append((ref, "head"))
513
for ref in repo.tags:
514
all_refs.append((ref, "tag"))
515
if os.path.isdir(os.path.join(server_repo_location, subpath)):
517
files = []
518
blobs = []
519
for entry in os.listdir(os.path.join(server_repo_location, subpath)):
521
if not os.path.basename(entry) == ".git":
522
files.append(os.path.join(subpath, entry))
523
infos = []
525
for file in files:
527
path = os.path.join(server_repo_location, file)
528
mimetype = guess_mime(path)
529
text = git_command(server_repo_location, None, "log", "--format='%H\n'",
531
shlex.quote(file)).decode()
532
sha = text.split("\n")[0]
534
identifier = f"/{username}/{repository}/{sha}"
535
last_commit = db.session.get(Commit, identifier)
537
info = {
539
"name": os.path.basename(file),
540
"serverPath": path,
541
"relativePath": file,
542
"link": os.path.join(f"/{username}/{repository}/tree/{branch}/", file),
543
"size": human_size(os.path.getsize(path)),
544
"mimetype": f"{mimetype}{f' ({mimetypes.guess_type(path)[1]})' if mimetypes.guess_type(path)[1] else ''}",
545
"commit": last_commit,
546
"shaSize": 7,
547
}
548
special_icon = config.match_icon(os.path.basename(file))
550
if special_icon:
551
info["icon"] = special_icon
552
elif os.path.isdir(path):
553
info["icon"] = config.folder_icon
554
elif mimetypes.guess_type(path)[0] in config.file_icons:
555
info["icon"] = config.file_icons[mimetypes.guess_type(path)[0]]
556
else:
557
info["icon"] = config.unknown_icon
558
if os.path.isdir(path):
560
infos.insert(0, info)
561
else:
562
infos.append(info)
563
return flask.render_template(
565
"repo-tree.html",
566
username=username,
567
repository=repository,
568
files=infos,
569
subpath=os.path.join("/", subpath),
570
branches=all_refs,
571
current=branch,
572
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
573
is_favourite=get_favourite(flask.session.get("username"), username, repository)
574
)
575
else:
576
path = os.path.join(server_repo_location, subpath)
577
if not os.path.exists(path):
579
return flask.render_template("not-found.html"), 404
580
mimetype = guess_mime(path)
582
mode = mimetype.split("/", 1)[0]
583
size = human_size(os.path.getsize(path))
584
special_icon = config.match_icon(os.path.basename(path))
586
if special_icon:
587
icon = special_icon
588
elif os.path.isdir(path):
589
icon = config.folder_icon
590
elif mimetypes.guess_type(path)[0] in config.file_icons:
591
icon = config.file_icons[mimetypes.guess_type(path)[0]]
592
else:
593
icon = config.unknown_icon
594
contents = None
596
if mode == "text":
597
contents = convert_to_html(path)
598
return flask.render_template(
600
"repo-file.html",
601
username=username,
602
repository=repository,
603
file=os.path.join(f"/{username}/{repository}/raw/{branch}/", subpath),
604
branches=all_refs,
605
current=branch,
606
mode=mode,
607
mimetype=mimetype,
608
detailedtype=magic.from_file(path),
609
size=size,
610
icon=icon,
611
subpath=os.path.join("/", subpath),
612
extension=pathlib.Path(path).suffix,
613
basename=os.path.basename(path),
614
contents=contents,
615
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
616
is_favourite=get_favourite(flask.session.get("username"), username, repository)
617
)
618
@repositories.route("/<username>/<repository>/commit/<sha>")
621
def repository_commit(username, repository, sha):
622
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
623
if not os.path.exists(server_repo_location):
624
app.logger.error(f"Cannot load {server_repo_location}")
625
flask.abort(404)
626
if not (get_visibility(username, repository) or get_permission_level(
627
flask.session.get("username"), username,
628
repository) is not None):
629
flask.abort(403)
630
app.logger.info(f"Loading {server_repo_location}")
632
if not os.path.exists(server_repo_location):
634
app.logger.error(f"Cannot load {server_repo_location}")
635
return flask.render_template("not-found.html"), 404
636
repo = git.Repo(server_repo_location)
638
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
639
files = git_command(os.path.join(server_repo_location, ".git"), None, "diff-tree", "-r",
641
"--name-only", "--no-commit-id", sha).decode().split("\n")[:-1]
642
print(files)
644
return flask.render_template(
646
"repo-commit.html",
647
username=username,
648
repository=repository,
649
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
650
is_favourite=get_favourite(flask.session.get("username"), username, repository),
651
diff={file: git_command(os.path.join(server_repo_location, ".git"), None, "diff",
652
str(sha) + "^!", "--", file).decode().split("\n") for
653
file in files},
654
data=db.session.get(Commit, f"/{username}/{repository}/{sha}"),
655
)
656
@repositories.route("/<username>/<repository>/forum/")
659
def repository_forum(username, repository):
660
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
661
if not os.path.exists(server_repo_location):
662
app.logger.error(f"Cannot load {server_repo_location}")
663
flask.abort(404)
664
if not (get_visibility(username, repository) or get_permission_level(
665
flask.session.get("username"), username,
666
repository) is not None):
667
flask.abort(403)
668
app.logger.info(f"Loading {server_repo_location}")
670
if not os.path.exists(server_repo_location):
672
app.logger.error(f"Cannot load {server_repo_location}")
673
return flask.render_template("not-found.html"), 404
674
repo = git.Repo(server_repo_location)
676
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
677
user = User.query.filter_by(username=flask.session.get("username")).first()
678
relationships = RepoAccess.query.filter_by(repo=repo_data)
679
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
680
return flask.render_template(
682
"repo-forum.html",
683
username=username,
684
repository=repository,
685
repo_data=repo_data,
686
relationships=relationships,
687
repo=repo,
688
user_relationship=user_relationship,
689
Post=Post,
690
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
691
is_favourite=get_favourite(flask.session.get("username"), username, repository),
692
default_branch=repo_data.default_branch
693
)
694
@repositories.route("/<username>/<repository>/forum/topic/<int:id>")
697
def repository_forum_topic(username, repository, id):
698
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
699
if not os.path.exists(server_repo_location):
700
app.logger.error(f"Cannot load {server_repo_location}")
701
flask.abort(404)
702
if not (get_visibility(username, repository) or get_permission_level(
703
flask.session.get("username"), username,
704
repository) is not None):
705
flask.abort(403)
706
app.logger.info(f"Loading {server_repo_location}")
708
if not os.path.exists(server_repo_location):
710
app.logger.error(f"Cannot load {server_repo_location}")
711
return flask.render_template("not-found.html"), 404
712
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
714
user = User.query.filter_by(username=flask.session.get("username")).first()
715
relationships = RepoAccess.query.filter_by(repo=repo_data)
716
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
717
post = Post.query.filter_by(id=id).first()
719
return flask.render_template(
721
"repo-topic.html",
722
username=username,
723
repository=repository,
724
repo_data=repo_data,
725
relationships=relationships,
726
user_relationship=user_relationship,
727
post=post,
728
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
729
is_favourite=get_favourite(flask.session.get("username"), username, repository),
730
default_branch=repo_data.default_branch
731
)
732
@repositories.route("/<username>/<repository>/forum/new", methods=["POST", "GET"])
735
def repository_forum_new(username, repository):
736
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
737
if not os.path.exists(server_repo_location):
738
app.logger.error(f"Cannot load {server_repo_location}")
739
flask.abort(404)
740
if not (get_visibility(username, repository) or get_permission_level(
741
flask.session.get("username"), username,
742
repository) is not None):
743
flask.abort(403)
744
app.logger.info(f"Loading {server_repo_location}")
746
if not os.path.exists(server_repo_location):
748
app.logger.error(f"Cannot load {server_repo_location}")
749
return flask.render_template("not-found.html"), 404
750
repo = git.Repo(server_repo_location)
752
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
753
user = User.query.filter_by(username=flask.session.get("username")).first()
754
relationships = RepoAccess.query.filter_by(repo=repo_data)
755
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
756
post = Post(user, repo_data, None, flask.request.form["subject"],
758
flask.request.form["message"])
759
db.session.add(post)
761
db.session.commit()
762
return flask.redirect(
764
flask.url_for(".repository_forum_thread", username=username, repository=repository,
765
post_id=post.number),
766
code=303)
767
@repositories.route("/<username>/<repository>/forum/<int:post_id>")
770
def repository_forum_thread(username, repository, post_id):
771
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
772
if not os.path.exists(server_repo_location):
773
app.logger.error(f"Cannot load {server_repo_location}")
774
flask.abort(404)
775
if not (get_visibility(username, repository) or get_permission_level(
776
flask.session.get("username"), username,
777
repository) is not None):
778
flask.abort(403)
779
app.logger.info(f"Loading {server_repo_location}")
781
if not os.path.exists(server_repo_location):
783
app.logger.error(f"Cannot load {server_repo_location}")
784
return flask.render_template("not-found.html"), 404
785
repo = git.Repo(server_repo_location)
787
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
788
user = User.query.filter_by(username=flask.session.get("username")).first()
789
relationships = RepoAccess.query.filter_by(repo=repo_data)
790
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
791
return flask.render_template(
793
"repo-forum-thread.html",
794
username=username,
795
repository=repository,
796
repo_data=repo_data,
797
relationships=relationships,
798
repo=repo,
799
Post=Post,
800
user_relationship=user_relationship,
801
post_id=post_id,
802
max_post_nesting=4,
803
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
804
is_favourite=get_favourite(flask.session.get("username"), username, repository),
805
parent=Post.query.filter_by(repo=repo_data, number=post_id).first(),
806
)
807
@repositories.route("/<username>/<repository>/forum/<int:post_id>/reply", methods=["POST"])
810
def repository_forum_reply(username, repository, post_id):
811
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
812
if not os.path.exists(server_repo_location):
813
app.logger.error(f"Cannot load {server_repo_location}")
814
flask.abort(404)
815
if not (get_visibility(username, repository) or get_permission_level(
816
flask.session.get("username"), username,
817
repository) is not None):
818
flask.abort(403)
819
app.logger.info(f"Loading {server_repo_location}")
821
if not os.path.exists(server_repo_location):
823
app.logger.error(f"Cannot load {server_repo_location}")
824
return flask.render_template("not-found.html"), 404
825
repo = git.Repo(server_repo_location)
827
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
828
user = User.query.filter_by(username=flask.session.get("username")).first()
829
relationships = RepoAccess.query.filter_by(repo=repo_data)
830
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
831
if not user:
832
flask.abort(401)
833
parent = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
835
post = Post(user, repo_data, parent, flask.request.form["subject"],
836
flask.request.form["message"])
837
db.session.add(post)
839
post.update_date()
840
db.session.commit()
841
return flask.redirect(
843
flask.url_for(".repository_forum_thread", username=username, repository=repository,
844
post_id=post_id),
845
code=303)
846
@repositories.route("/<username>/<repository>/forum/<int:post_id>/voteup",
849
defaults={"score": 1})
850
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votedown",
851
defaults={"score": -1})
852
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votes", defaults={"score": 0})
853
def repository_forum_vote(username, repository, post_id, score):
854
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
855
if not os.path.exists(server_repo_location):
856
app.logger.error(f"Cannot load {server_repo_location}")
857
flask.abort(404)
858
if not (get_visibility(username, repository) or get_permission_level(
859
flask.session.get("username"), username,
860
repository) is not None):
861
flask.abort(403)
862
app.logger.info(f"Loading {server_repo_location}")
864
if not os.path.exists(server_repo_location):
866
app.logger.error(f"Cannot load {server_repo_location}")
867
return flask.render_template("not-found.html"), 404
868
repo = git.Repo(server_repo_location)
870
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
871
user = User.query.filter_by(username=flask.session.get("username")).first()
872
relationships = RepoAccess.query.filter_by(repo=repo_data)
873
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
874
if not user:
875
flask.abort(401)
876
post = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
878
if score:
880
old_relationship = PostVote.query.filter_by(user_username=user.username,
881
post_identifier=post.identifier).first()
882
if old_relationship:
883
if score == old_relationship.vote_score:
884
db.session.delete(old_relationship)
885
post.vote_sum -= old_relationship.vote_score
886
else:
887
post.vote_sum -= old_relationship.vote_score
888
post.vote_sum += score
889
old_relationship.vote_score = score
890
else:
891
relationship = PostVote(user, post, score)
892
post.vote_sum += score
893
db.session.add(relationship)
894
db.session.commit()
896
user_vote = PostVote.query.filter_by(user_username=user.username,
898
post_identifier=post.identifier).first()
899
response = flask.make_response(
900
str(post.vote_sum) + " " + str(user_vote.vote_score if user_vote else 0))
901
response.content_type = "text/plain"
902
return response
904
@repositories.route("/<username>/<repository>/favourite")
907
def repository_favourite(username, repository):
908
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
909
if not os.path.exists(server_repo_location):
910
app.logger.error(f"Cannot load {server_repo_location}")
911
flask.abort(404)
912
if not (get_visibility(username, repository) or get_permission_level(
913
flask.session.get("username"), username,
914
repository) is not None):
915
flask.abort(403)
916
app.logger.info(f"Loading {server_repo_location}")
918
if not os.path.exists(server_repo_location):
920
app.logger.error(f"Cannot load {server_repo_location}")
921
return flask.render_template("not-found.html"), 404
922
repo = git.Repo(server_repo_location)
924
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
925
user = User.query.filter_by(username=flask.session.get("username")).first()
926
relationships = RepoAccess.query.filter_by(repo=repo_data)
927
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
928
if not user:
929
flask.abort(401)
930
old_relationship = RepoFavourite.query.filter_by(user_username=user.username,
932
repo_route=repo_data.route).first()
933
if old_relationship:
934
db.session.delete(old_relationship)
935
else:
936
relationship = RepoFavourite(user, repo_data)
937
db.session.add(relationship)
938
db.session.commit()
940
return flask.redirect(flask.url_for("favourites"), code=303)
942
@repositories.route("/<username>/<repository>/users/", methods=["GET", "POST"])
945
def repository_users(username, repository):
946
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
947
if not os.path.exists(server_repo_location):
948
app.logger.error(f"Cannot load {server_repo_location}")
949
flask.abort(404)
950
if not (get_visibility(username, repository) or get_permission_level(
951
flask.session.get("username"), username,
952
repository) is not None):
953
flask.abort(403)
954
app.logger.info(f"Loading {server_repo_location}")
956
if not os.path.exists(server_repo_location):
958
app.logger.error(f"Cannot load {server_repo_location}")
959
return flask.render_template("not-found.html"), 404
960
repo = git.Repo(server_repo_location)
962
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
963
user = User.query.filter_by(username=flask.session.get("username")).first()
964
relationships = RepoAccess.query.filter_by(repo=repo_data)
965
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
966
if flask.request.method == "GET":
968
return flask.render_template(
969
"repo-users.html",
970
username=username,
971
repository=repository,
972
repo_data=repo_data,
973
relationships=relationships,
974
repo=repo,
975
user_relationship=user_relationship,
976
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
977
is_favourite=get_favourite(flask.session.get("username"), username, repository)
978
)
979
else:
980
if get_permission_level(flask.session.get("username"), username, repository) != 2:
981
flask.abort(401)
982
if flask.request.form.get("new-username"):
984
# Create new relationship
985
new_user = User.query.filter_by(
986
username=flask.request.form.get("new-username")).first()
987
relationship = RepoAccess(new_user, repo_data, flask.request.form.get("new-level"))
988
db.session.add(relationship)
989
db.session.commit()
990
if flask.request.form.get("update-username"):
991
# Create new relationship
992
updated_user = User.query.filter_by(
993
username=flask.request.form.get("update-username")).first()
994
relationship = RepoAccess.query.filter_by(repo=repo_data, user=updated_user).first()
995
if flask.request.form.get("update-level") == -1:
996
relationship.delete()
997
else:
998
relationship.access_level = flask.request.form.get("update-level")
999
db.session.commit()
1000
return flask.redirect(
1002
app.url_for(".repository_users", username=username, repository=repository))
1003
@repositories.route("/<username>/<repository>/branches/")
1006
def repository_branches(username, repository):
1007
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1008
if not os.path.exists(server_repo_location):
1009
app.logger.error(f"Cannot load {server_repo_location}")
1010
flask.abort(404)
1011
if not (get_visibility(username, repository) or get_permission_level(
1012
flask.session.get("username"), username,
1013
repository) is not None):
1014
flask.abort(403)
1015
app.logger.info(f"Loading {server_repo_location}")
1017
if not os.path.exists(server_repo_location):
1019
app.logger.error(f"Cannot load {server_repo_location}")
1020
return flask.render_template("not-found.html"), 404
1021
repo = git.Repo(server_repo_location)
1023
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1024
return flask.render_template(
1026
"repo-branches.html",
1027
username=username,
1028
repository=repository,
1029
repo_data=repo_data,
1030
repo=repo,
1031
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1032
is_favourite=get_favourite(flask.session.get("username"), username, repository)
1033
)
1034
@repositories.route("/<username>/<repository>/log/", defaults={"branch": None})
1037
@repositories.route("/<username>/<repository>/log/<branch>/")
1038
def repository_log(username, repository, branch):
1039
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1040
if not os.path.exists(server_repo_location):
1041
app.logger.error(f"Cannot load {server_repo_location}")
1042
flask.abort(404)
1043
if not (get_visibility(username, repository) or get_permission_level(
1044
flask.session.get("username"), username,
1045
repository) is not None):
1046
flask.abort(403)
1047
app.logger.info(f"Loading {server_repo_location}")
1049
if not os.path.exists(server_repo_location):
1051
app.logger.error(f"Cannot load {server_repo_location}")
1052
return flask.render_template("not-found.html"), 404
1053
repo = git.Repo(server_repo_location)
1055
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1056
if not repo_data.default_branch:
1057
if repo.heads:
1058
repo_data.default_branch = repo.heads[0].name
1059
else:
1060
return flask.render_template("empty.html",
1061
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
1062
if not branch:
1063
branch = repo_data.default_branch
1064
return flask.redirect(f"./{branch}", code=302)
1065
if branch.startswith("tag:"):
1067
ref = f"tags/{branch[4:]}"
1068
elif branch.startswith("~"):
1069
ref = branch[1:]
1070
else:
1071
ref = f"heads/{branch}"
1072
ref = ref.replace("~", "/") # encode slashes for URL support
1074
try:
1076
repo.git.checkout("-f", ref)
1077
except git.exc.GitCommandError:
1078
return flask.render_template("not-found.html"), 404
1079
branches = repo.heads
1081
all_refs = []
1083
for ref in repo.heads:
1084
all_refs.append((ref, "head"))
1085
for ref in repo.tags:
1086
all_refs.append((ref, "tag"))
1087
commit_list = [f"/{username}/{repository}/{sha}" for sha in
1089
git_command(server_repo_location, None, "log",
1090
"--format='%H'").decode().split("\n")]
1091
commits = Commit.query.filter(Commit.identifier.in_(commit_list))
1093
return flask.render_template(
1095
"repo-log.html",
1096
username=username,
1097
repository=repository,
1098
branches=all_refs,
1099
current=branch,
1100
repo_data=repo_data,
1101
repo=repo,
1102
commits=commits,
1103
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1104
is_favourite=get_favourite(flask.session.get("username"), username, repository)
1105
)
1106
@repositories.route("/<username>/<repository>/prs/", methods=["GET", "POST"])
1109
def repository_prs(username, repository):
1110
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1111
if not os.path.exists(server_repo_location):
1112
app.logger.error(f"Cannot load {server_repo_location}")
1113
flask.abort(404)
1114
if not (get_visibility(username, repository) or get_permission_level(
1115
flask.session.get("username"), username,
1116
repository) is not None):
1117
flask.abort(403)
1118
app.logger.info(f"Loading {server_repo_location}")
1120
if not os.path.exists(server_repo_location):
1122
app.logger.error(f"Cannot load {server_repo_location}")
1123
return flask.render_template("not-found.html"), 404
1124
if flask.request.method == "GET":
1126
repo = git.Repo(server_repo_location)
1127
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1128
user = User.query.filter_by(username=flask.session.get("username")).first()
1129
return flask.render_template(
1131
"repo-prs.html",
1132
username=username,
1133
repository=repository,
1134
repo_data=repo_data,
1135
repo=repo,
1136
PullRequest=PullRequest,
1137
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1138
is_favourite=get_favourite(flask.session.get("username"), username, repository),
1139
default_branch=repo_data.default_branch,
1140
branches=repo.branches
1141
)
1142
else:
1144
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1145
head = flask.request.form.get("head")
1146
head_route = flask.request.form.get("headroute")
1147
base = flask.request.form.get("base")
1148
if not head and base and head_route:
1150
return flask.redirect(".", 400)
1151
head_repo = git.Repo(os.path.join(config.REPOS_PATH, head_route.lstrip("/")))
1153
base_repo = git.Repo(server_repo_location)
1154
print(head_repo)
1155
if head not in head_repo.branches or base not in base_repo.branches:
1157
flask.flash(Markup(
1158
"<iconify-icon icon='mdi:error'></iconify-icon>" + _("Bad branch name")),
1159
category="error")
1160
return flask.redirect(".", 303)
1161
head_data = db.session.get(Repo, head_route)
1163
if not head_data.visibility:
1164
flask.flash(Markup(
1165
"<iconify-icon icon='mdi:error'></iconify-icon>" + _(
1166
"Head can't be restricted")),
1167
category="error")
1168
return flask.redirect(".", 303)
1169
pull_request = PullRequest(repo_data, head, head_data, base,
1171
db.session.get(User, flask.session["username"]))
1172
db.session.add(pull_request)
1174
db.session.commit()
1175
return flask.redirect(".", 303)
1177
@repositories.route("/<username>/<repository>/prs/merge", methods=["POST"])
1180
def repository_prs_merge(username, repository):
1181
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1182
if not os.path.exists(server_repo_location):
1183
app.logger.error(f"Cannot load {server_repo_location}")
1184
flask.abort(404)
1185
if not (get_visibility(username, repository) or get_permission_level(
1186
flask.session.get("username"), username,
1187
repository) is not None):
1188
flask.abort(403)
1189
if not get_permission_level(flask.session.get("username"), username, repository):
1191
flask.abort(401)
1192
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1194
repo = git.Repo(server_repo_location)
1195
id = flask.request.form.get("id")
1196
pull_request = db.session.get(PullRequest, id)
1198
if pull_request:
1200
result = celery_tasks.merge_heads.delay(
1201
pull_request.head_route,
1202
pull_request.head_branch,
1203
pull_request.base_route,
1204
pull_request.base_branch,
1205
simulate=True
1206
)
1207
task_result = worker.AsyncResult(result.id)
1208
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303)
1210
# db.session.delete(pull_request)
1211
# db.session.commit()
1212
else:
1213
flask.abort(400)
1214
@repositories.route("/<username>/<repository>/prs/<int:id>/merge")
1217
def repository_prs_merge_stage_two(username, repository, id):
1218
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1219
if not os.path.exists(server_repo_location):
1220
app.logger.error(f"Cannot load {server_repo_location}")
1221
flask.abort(404)
1222
if not (get_visibility(username, repository) or get_permission_level(
1223
flask.session.get("username"), username,
1224
repository) is not None):
1225
flask.abort(403)
1226
if not get_permission_level(flask.session.get("username"), username, repository):
1228
flask.abort(401)
1229
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1231
repo = git.Repo(server_repo_location)
1232
pull_request = db.session.get(PullRequest, id)
1234
if pull_request:
1236
result = celery_tasks.merge_heads.delay(
1237
pull_request.head_route,
1238
pull_request.head_branch,
1239
pull_request.base_route,
1240
pull_request.base_branch,
1241
simulate=False
1242
)
1243
task_result = worker.AsyncResult(result.id)
1244
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303)
1246
# db.session.delete(pull_request)
1247
# db.session.commit()
1248
else:
1249
flask.abort(400)
1250
@app.route("/task/<task_id>")
1253
def task_monitor(task_id):
1254
task_result = worker.AsyncResult(task_id)
1255
print(task_result.status)
1256
return flask.render_template("task-monitor.html", result=task_result)
1258
@repositories.route("/<username>/<repository>/prs/delete", methods=["POST"])
1261
def repository_prs_delete(username, repository):
1262
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1263
if not os.path.exists(server_repo_location):
1264
app.logger.error(f"Cannot load {server_repo_location}")
1265
flask.abort(404)
1266
if not (get_visibility(username, repository) or get_permission_level(
1267
flask.session.get("username"), username,
1268
repository) is not None):
1269
flask.abort(403)
1270
if not get_permission_level(flask.session.get("username"), username, repository):
1272
flask.abort(401)
1273
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1275
repo = git.Repo(server_repo_location)
1276
id = flask.request.form.get("id")
1277
pull_request = db.session.get(PullRequest, id)
1279
if pull_request:
1281
db.session.delete(pull_request)
1282
db.session.commit()
1283
return flask.redirect(".", 303)
1285
@repositories.route("/<username>/<repository>/settings/")
1288
def repository_settings(username, repository):
1289
if get_permission_level(flask.session.get("username"), username, repository) != 2:
1290
flask.abort(401)
1291
return flask.render_template("repo-settings.html", username=username, repository=repository)
1293
@app.errorhandler(404)
1296
def e404(error):
1297
return flask.render_template("not-found.html"), 404
1298
@app.errorhandler(401)
1301
def e401(error):
1302
return flask.render_template("unauthorised.html"), 401
1303
@app.errorhandler(403)
1306
def e403(error):
1307
return flask.render_template("forbidden.html"), 403
1308
@app.errorhandler(418)
1311
def e418(error):
1312
return flask.render_template("teapot.html"), 418
1313
@app.errorhandler(405)
1316
def e405(error):
1317
return flask.render_template("method-not-allowed.html"), 405
1318
if __name__ == "__main__":
1321
app.run(debug=True, port=8080, host="0.0.0.0")
1322
app.register_blueprint(repositories)