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("/notifications/mark-all-read", methods=["POST"])
228
def mark_all_read():
229
if not flask.session.get("username"):
230
flask.abort(401)
231
notifications = UserNotification.query.filter_by(
233
user_username=flask.session.get("username"))
234
for notification in notifications:
235
notification.mark_read()
236
db.session.commit()
237
return flask.redirect("/notifications/", code=303)
238
@app.route("/accounts/", methods=["GET", "POST"])
241
def login():
242
if flask.request.method == "GET":
243
return flask.render_template("login.html")
244
else:
245
if "login" in flask.request.form:
246
username = flask.request.form["username"]
247
password = flask.request.form["password"]
248
user = User.query.filter_by(username=username).first()
250
if user and bcrypt.check_password_hash(user.password_hashed, password):
252
flask.session["username"] = user.username
253
flask.flash(
254
Markup("<iconify-icon icon='mdi:account'></iconify-icon>" + _(
255
"Successfully logged in as {username}").format(username=username)),
256
category="success")
257
return flask.redirect("/", code=303)
258
elif not user:
259
flask.flash(Markup(
260
"<iconify-icon icon='mdi:account-question'></iconify-icon>" + _(
261
"User not found")),
262
category="alert")
263
return flask.render_template("login.html")
264
else:
265
flask.flash(Markup(
266
"<iconify-icon icon='mdi:account-question'></iconify-icon>" + _(
267
"Invalid password")),
268
category="error")
269
return flask.render_template("login.html")
270
if "signup" in flask.request.form:
271
username = flask.request.form["username"]
272
password = flask.request.form["password"]
273
password2 = flask.request.form["password2"]
274
email = flask.request.form.get("email")
275
email2 = flask.request.form.get("email2") # repeat email is a honeypot
276
name = flask.request.form.get("name")
277
if not only_chars(username,
279
"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"):
280
flask.flash(Markup(
281
_("Usernames may only contain Latin alphabet, numbers, '-' and '_'")),
282
category="error")
283
return flask.render_template("login.html")
284
if username in config.RESERVED_NAMES:
286
flask.flash(
287
Markup(
288
"<iconify-icon icon='mdi:account-error'></iconify-icon>" + _(
289
"Sorry, {username} is a system path").format(
290
username=username)),
291
category="error")
292
return flask.render_template("login.html")
293
user_check = User.query.filter_by(username=username).first()
295
if user_check or email2: # make the honeypot look like a normal error
296
flask.flash(
297
Markup(
298
"<iconify-icon icon='mdi:account-error'></iconify-icon>" + _(
299
"The username {username} is taken").format(
300
username=username)),
301
category="error")
302
return flask.render_template("login.html")
303
if password2 != password:
305
flask.flash(Markup("<iconify-icon icon='mdi:key-alert'></iconify-icon>" + _(
306
"Make sure the passwords match")),
307
category="error")
308
return flask.render_template("login.html")
309
user = User(username, password, email, name)
311
db.session.add(user)
312
db.session.commit()
313
flask.session["username"] = user.username
314
flask.flash(Markup(
315
"<iconify-icon icon='mdi:account'></iconify-icon>" + _(
316
"Successfully created and logged in as {username}").format(
317
username=username)),
318
category="success")
319
notification = Notification({"type": "welcome"})
321
db.session.add(notification)
322
db.session.commit()
323
result = celery_tasks.send_notification.delay(notification.id, [username], 1)
325
return flask.redirect("/", code=303)
327
@app.route("/newrepo/", methods=["GET", "POST"])
330
def new_repo():
331
if not flask.session.get("username"):
332
flask.abort(401)
333
if flask.request.method == "GET":
334
return flask.render_template("new-repo.html")
335
else:
336
name = flask.request.form["name"]
337
visibility = int(flask.request.form["visibility"])
338
if not only_chars(name,
340
"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"):
341
flask.flash(Markup(
342
"<iconify-icon icon='mdi:error'></iconify-icon>" + _(
343
"Repository names may only contain Latin alphabet, numbers, '-' and '_'")),
344
category="error")
345
return flask.render_template("new-repo.html")
346
user = User.query.filter_by(username=flask.session.get("username")).first()
348
repo = Repo(user, name, visibility)
350
db.session.add(repo)
351
db.session.commit()
352
flask.flash(Markup(_("Successfully created repository {name}").format(name=name)),
354
category="success")
355
return flask.redirect(repo.route, code=303)
356
@app.route("/logout")
359
def logout():
360
flask.session.clear()
361
flask.flash(Markup(
362
"<iconify-icon icon='mdi:account'></iconify-icon>" + _("Successfully logged out")),
363
category="info")
364
return flask.redirect("/", code=303)
365
@app.route("/<username>/", methods=["GET", "POST"])
368
def user_profile(username):
369
old_relationship = UserFollow.query.filter_by(
370
follower_username=flask.session.get("username"),
371
followed_username=username).first()
372
if flask.request.method == "GET":
373
user = User.query.filter_by(username=username).first()
374
match flask.request.args.get("action"):
375
case "repositories":
376
repos = Repo.query.filter_by(owner_name=username, visibility=2)
377
return flask.render_template("user-profile-repositories.html", user=user,
378
repos=repos,
379
relationship=old_relationship)
380
case "followers":
381
return flask.render_template("user-profile-followers.html", user=user,
382
relationship=old_relationship)
383
case "follows":
384
return flask.render_template("user-profile-follows.html", user=user,
385
relationship=old_relationship)
386
case _:
387
return flask.render_template("user-profile-overview.html", user=user,
388
relationship=old_relationship)
389
elif flask.request.method == "POST":
391
match flask.request.args.get("action"):
392
case "follow":
393
if username == flask.session.get("username"):
394
flask.abort(403)
395
if old_relationship:
396
db.session.delete(old_relationship)
397
else:
398
relationship = UserFollow(
399
flask.session.get("username"),
400
username
401
)
402
db.session.add(relationship)
403
db.session.commit()
404
user = db.session.get(User, username)
406
author = db.session.get(User, flask.session.get("username"))
407
notification = Notification({"type": "update", "version": "0.0.0"})
408
db.session.add(notification)
409
db.session.commit()
410
result = celery_tasks.send_notification.delay(notification.id, [username],
412
1)
413
db.session.commit()
415
return flask.redirect("?", code=303)
416
@app.route("/<username>/<repository>/")
419
def repository_index(username, repository):
420
return flask.redirect("./tree", code=302)
421
@app.route("/info/<username>/avatar")
424
def user_avatar(username):
425
serverUserdataLocation = os.path.join(config.USERDATA_PATH, username)
426
if not os.path.exists(serverUserdataLocation):
428
return flask.render_template("not-found.html"), 404
429
return flask.send_from_directory(serverUserdataLocation, "avatar.png")
431
@app.route("/<username>/<repository>/raw/<branch>/<path:subpath>")
434
def repository_raw(username, repository, branch, subpath):
435
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
436
if not os.path.exists(server_repo_location):
437
app.logger.error(f"Cannot load {server_repo_location}")
438
flask.abort(404)
439
if not (get_visibility(username, repository) or get_permission_level(
440
flask.session.get("username"), username,
441
repository) is not None):
442
flask.abort(403)
443
app.logger.info(f"Loading {server_repo_location}")
445
if not os.path.exists(server_repo_location):
447
app.logger.error(f"Cannot load {server_repo_location}")
448
return flask.render_template("not-found.html"), 404
449
repo = git.Repo(server_repo_location)
451
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
452
if not repo_data.default_branch:
453
if repo.heads:
454
repo_data.default_branch = repo.heads[0].name
455
else:
456
return flask.render_template("empty.html",
457
remote=f"http://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
458
if not branch:
459
branch = repo_data.default_branch
460
return flask.redirect(f"./{branch}", code=302)
461
if branch.startswith("tag:"):
463
ref = f"tags/{branch[4:]}"
464
elif branch.startswith("~"):
465
ref = branch[1:]
466
else:
467
ref = f"heads/{branch}"
468
ref = ref.replace("~", "/") # encode slashes for URL support
470
try:
472
repo.git.checkout("-f", ref)
473
except git.exc.GitCommandError:
474
return flask.render_template("not-found.html"), 404
475
return flask.send_from_directory(config.REPOS_PATH,
477
os.path.join(username, repository, subpath))
478
@repositories.route("/<username>/<repository>/tree/", defaults={"branch": None, "subpath": ""})
481
@repositories.route("/<username>/<repository>/tree/<branch>/", defaults={"subpath": ""})
482
@repositories.route("/<username>/<repository>/tree/<branch>/<path:subpath>")
483
def repository_tree(username, repository, branch, subpath):
484
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
485
if not os.path.exists(server_repo_location):
486
app.logger.error(f"Cannot load {server_repo_location}")
487
flask.abort(404)
488
if not (get_visibility(username, repository) or get_permission_level(
489
flask.session.get("username"), username,
490
repository) is not None):
491
flask.abort(403)
492
app.logger.info(f"Loading {server_repo_location}")
494
repo = git.Repo(server_repo_location)
496
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
497
if not repo_data.default_branch:
498
if repo.heads:
499
repo_data.default_branch = repo.heads[0].name
500
else:
501
return flask.render_template("empty.html",
502
remote=f"{config.www_protocol}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
503
if not branch:
504
branch = repo_data.default_branch
505
return flask.redirect(f"./{branch}", code=302)
506
if branch.startswith("tag:"):
508
ref = f"tags/{branch[4:]}"
509
elif branch.startswith("~"):
510
ref = branch[1:]
511
else:
512
ref = f"heads/{branch}"
513
ref = ref.replace("~", "/") # encode slashes for URL support
515
try:
517
repo.git.checkout("-f", ref)
518
except git.exc.GitCommandError:
519
return flask.render_template("not-found.html"), 404
520
branches = repo.heads
522
all_refs = []
524
for ref in repo.heads:
525
all_refs.append((ref, "head"))
526
for ref in repo.tags:
527
all_refs.append((ref, "tag"))
528
if os.path.isdir(os.path.join(server_repo_location, subpath)):
530
files = []
531
blobs = []
532
for entry in os.listdir(os.path.join(server_repo_location, subpath)):
534
if not os.path.basename(entry) == ".git":
535
files.append(os.path.join(subpath, entry))
536
infos = []
538
for file in files:
540
path = os.path.join(server_repo_location, file)
541
mimetype = guess_mime(path)
542
text = git_command(server_repo_location, None, "log", "--format='%H\n'",
544
shlex.quote(file)).decode()
545
sha = text.split("\n")[0]
547
identifier = f"/{username}/{repository}/{sha}"
548
last_commit = db.session.get(Commit, identifier)
550
info = {
552
"name": os.path.basename(file),
553
"serverPath": path,
554
"relativePath": file,
555
"link": os.path.join(f"/{username}/{repository}/tree/{branch}/", file),
556
"size": human_size(os.path.getsize(path)),
557
"mimetype": f"{mimetype}{f' ({mimetypes.guess_type(path)[1]})' if mimetypes.guess_type(path)[1] else ''}",
558
"commit": last_commit,
559
"shaSize": 7,
560
}
561
special_icon = config.match_icon(os.path.basename(file))
563
if special_icon:
564
info["icon"] = special_icon
565
elif os.path.isdir(path):
566
info["icon"] = config.folder_icon
567
elif mimetypes.guess_type(path)[0] in config.file_icons:
568
info["icon"] = config.file_icons[mimetypes.guess_type(path)[0]]
569
else:
570
info["icon"] = config.unknown_icon
571
if os.path.isdir(path):
573
infos.insert(0, info)
574
else:
575
infos.append(info)
576
return flask.render_template(
578
"repo-tree.html",
579
username=username,
580
repository=repository,
581
files=infos,
582
subpath=os.path.join("/", subpath),
583
branches=all_refs,
584
current=branch,
585
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
586
is_favourite=get_favourite(flask.session.get("username"), username, repository)
587
)
588
else:
589
path = os.path.join(server_repo_location, subpath)
590
if not os.path.exists(path):
592
return flask.render_template("not-found.html"), 404
593
mimetype = guess_mime(path)
595
mode = mimetype.split("/", 1)[0]
596
size = human_size(os.path.getsize(path))
597
special_icon = config.match_icon(os.path.basename(path))
599
if special_icon:
600
icon = special_icon
601
elif os.path.isdir(path):
602
icon = config.folder_icon
603
elif mimetypes.guess_type(path)[0] in config.file_icons:
604
icon = config.file_icons[mimetypes.guess_type(path)[0]]
605
else:
606
icon = config.unknown_icon
607
contents = None
609
if mode == "text":
610
contents = convert_to_html(path)
611
return flask.render_template(
613
"repo-file.html",
614
username=username,
615
repository=repository,
616
file=os.path.join(f"/{username}/{repository}/raw/{branch}/", subpath),
617
branches=all_refs,
618
current=branch,
619
mode=mode,
620
mimetype=mimetype,
621
detailedtype=magic.from_file(path),
622
size=size,
623
icon=icon,
624
subpath=os.path.join("/", subpath),
625
extension=pathlib.Path(path).suffix,
626
basename=os.path.basename(path),
627
contents=contents,
628
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
629
is_favourite=get_favourite(flask.session.get("username"), username, repository)
630
)
631
@repositories.route("/<username>/<repository>/commit/<sha>")
634
def repository_commit(username, repository, sha):
635
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
636
if not os.path.exists(server_repo_location):
637
app.logger.error(f"Cannot load {server_repo_location}")
638
flask.abort(404)
639
if not (get_visibility(username, repository) or get_permission_level(
640
flask.session.get("username"), username,
641
repository) is not None):
642
flask.abort(403)
643
app.logger.info(f"Loading {server_repo_location}")
645
if not os.path.exists(server_repo_location):
647
app.logger.error(f"Cannot load {server_repo_location}")
648
return flask.render_template("not-found.html"), 404
649
repo = git.Repo(server_repo_location)
651
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
652
files = git_command(os.path.join(server_repo_location, ".git"), None, "diff-tree", "-r",
654
"--name-only", "--no-commit-id", sha).decode().split("\n")[:-1]
655
print(files)
657
return flask.render_template(
659
"repo-commit.html",
660
username=username,
661
repository=repository,
662
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
663
is_favourite=get_favourite(flask.session.get("username"), username, repository),
664
diff={file: git_command(os.path.join(server_repo_location, ".git"), None, "diff",
665
str(sha) + "^!", "--", file).decode().split("\n") for
666
file in files},
667
data=db.session.get(Commit, f"/{username}/{repository}/{sha}"),
668
)
669
@repositories.route("/<username>/<repository>/forum/")
672
def repository_forum(username, repository):
673
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
674
if not os.path.exists(server_repo_location):
675
app.logger.error(f"Cannot load {server_repo_location}")
676
flask.abort(404)
677
if not (get_visibility(username, repository) or get_permission_level(
678
flask.session.get("username"), username,
679
repository) is not None):
680
flask.abort(403)
681
app.logger.info(f"Loading {server_repo_location}")
683
if not os.path.exists(server_repo_location):
685
app.logger.error(f"Cannot load {server_repo_location}")
686
return flask.render_template("not-found.html"), 404
687
repo = git.Repo(server_repo_location)
689
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
690
user = User.query.filter_by(username=flask.session.get("username")).first()
691
relationships = RepoAccess.query.filter_by(repo=repo_data)
692
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
693
return flask.render_template(
695
"repo-forum.html",
696
username=username,
697
repository=repository,
698
repo_data=repo_data,
699
relationships=relationships,
700
repo=repo,
701
user_relationship=user_relationship,
702
Post=Post,
703
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
704
is_favourite=get_favourite(flask.session.get("username"), username, repository),
705
default_branch=repo_data.default_branch
706
)
707
@repositories.route("/<username>/<repository>/forum/topic/<int:id>")
710
def repository_forum_topic(username, repository, id):
711
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
712
if not os.path.exists(server_repo_location):
713
app.logger.error(f"Cannot load {server_repo_location}")
714
flask.abort(404)
715
if not (get_visibility(username, repository) or get_permission_level(
716
flask.session.get("username"), username,
717
repository) is not None):
718
flask.abort(403)
719
app.logger.info(f"Loading {server_repo_location}")
721
if not os.path.exists(server_repo_location):
723
app.logger.error(f"Cannot load {server_repo_location}")
724
return flask.render_template("not-found.html"), 404
725
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
727
user = User.query.filter_by(username=flask.session.get("username")).first()
728
relationships = RepoAccess.query.filter_by(repo=repo_data)
729
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
730
post = Post.query.filter_by(id=id).first()
732
return flask.render_template(
734
"repo-topic.html",
735
username=username,
736
repository=repository,
737
repo_data=repo_data,
738
relationships=relationships,
739
user_relationship=user_relationship,
740
post=post,
741
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
742
is_favourite=get_favourite(flask.session.get("username"), username, repository),
743
default_branch=repo_data.default_branch
744
)
745
@repositories.route("/<username>/<repository>/forum/new", methods=["POST", "GET"])
748
def repository_forum_new(username, repository):
749
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
750
if not os.path.exists(server_repo_location):
751
app.logger.error(f"Cannot load {server_repo_location}")
752
flask.abort(404)
753
if not (get_visibility(username, repository) or get_permission_level(
754
flask.session.get("username"), username,
755
repository) is not None):
756
flask.abort(403)
757
app.logger.info(f"Loading {server_repo_location}")
759
if not os.path.exists(server_repo_location):
761
app.logger.error(f"Cannot load {server_repo_location}")
762
return flask.render_template("not-found.html"), 404
763
repo = git.Repo(server_repo_location)
765
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
766
user = User.query.filter_by(username=flask.session.get("username")).first()
767
relationships = RepoAccess.query.filter_by(repo=repo_data)
768
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
769
post = Post(user, repo_data, None, flask.request.form["subject"],
771
flask.request.form["message"])
772
db.session.add(post)
774
db.session.commit()
775
return flask.redirect(
777
flask.url_for(".repository_forum_thread", username=username, repository=repository,
778
post_id=post.number),
779
code=303)
780
@repositories.route("/<username>/<repository>/forum/<int:post_id>")
783
def repository_forum_thread(username, repository, post_id):
784
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
785
if not os.path.exists(server_repo_location):
786
app.logger.error(f"Cannot load {server_repo_location}")
787
flask.abort(404)
788
if not (get_visibility(username, repository) or get_permission_level(
789
flask.session.get("username"), username,
790
repository) is not None):
791
flask.abort(403)
792
app.logger.info(f"Loading {server_repo_location}")
794
if not os.path.exists(server_repo_location):
796
app.logger.error(f"Cannot load {server_repo_location}")
797
return flask.render_template("not-found.html"), 404
798
repo = git.Repo(server_repo_location)
800
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
801
user = User.query.filter_by(username=flask.session.get("username")).first()
802
relationships = RepoAccess.query.filter_by(repo=repo_data)
803
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
804
return flask.render_template(
806
"repo-forum-thread.html",
807
username=username,
808
repository=repository,
809
repo_data=repo_data,
810
relationships=relationships,
811
repo=repo,
812
Post=Post,
813
user_relationship=user_relationship,
814
post_id=post_id,
815
max_post_nesting=4,
816
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
817
is_favourite=get_favourite(flask.session.get("username"), username, repository),
818
parent=Post.query.filter_by(repo=repo_data, number=post_id).first(),
819
)
820
@repositories.route("/<username>/<repository>/forum/<int:post_id>/change-state", methods=["POST"])
823
def repository_forum_change_state(username, repository, post_id):
824
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
825
if not os.path.exists(server_repo_location):
826
app.logger.error(f"Cannot load {server_repo_location}")
827
flask.abort(404)
828
if not (get_visibility(username, repository) or get_permission_level(
829
flask.session.get("username"), username,
830
repository) is not None):
831
flask.abort(403)
832
app.logger.info(f"Loading {server_repo_location}")
834
repo = git.Repo(server_repo_location)
836
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
837
user = User.query.filter_by(username=flask.session.get("username")).first()
838
relationships = RepoAccess.query.filter_by(repo=repo_data)
839
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
840
post = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
842
if not post:
844
flask.abort(404)
845
post.state = int(flask.request.form["new-state"])
847
db.session.commit()
849
return flask.redirect(
851
flask.url_for(".repository_forum_thread", username=username, repository=repository,
852
post_id=post_id),
853
code=303)
854
@repositories.route("/<username>/<repository>/forum/<int:post_id>/reply", methods=["POST"])
857
def repository_forum_reply(username, repository, post_id):
858
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
859
if not os.path.exists(server_repo_location):
860
app.logger.error(f"Cannot load {server_repo_location}")
861
flask.abort(404)
862
if not (get_visibility(username, repository) or get_permission_level(
863
flask.session.get("username"), username,
864
repository) is not None):
865
flask.abort(403)
866
app.logger.info(f"Loading {server_repo_location}")
868
if not os.path.exists(server_repo_location):
870
app.logger.error(f"Cannot load {server_repo_location}")
871
return flask.render_template("not-found.html"), 404
872
repo = git.Repo(server_repo_location)
874
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
875
user = User.query.filter_by(username=flask.session.get("username")).first()
876
relationships = RepoAccess.query.filter_by(repo=repo_data)
877
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
878
if not user:
879
flask.abort(401)
880
parent = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
882
post = Post(user, repo_data, parent, flask.request.form["subject"],
883
flask.request.form["message"])
884
db.session.add(post)
886
post.update_date()
887
db.session.commit()
888
return flask.redirect(
890
flask.url_for(".repository_forum_thread", username=username, repository=repository,
891
post_id=post_id),
892
code=303)
893
@repositories.route("/<username>/<repository>/forum/<int:post_id>/voteup",
896
defaults={"score": 1})
897
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votedown",
898
defaults={"score": -1})
899
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votes", defaults={"score": 0})
900
def repository_forum_vote(username, repository, post_id, score):
901
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
902
if not os.path.exists(server_repo_location):
903
app.logger.error(f"Cannot load {server_repo_location}")
904
flask.abort(404)
905
if not (get_visibility(username, repository) or get_permission_level(
906
flask.session.get("username"), username,
907
repository) is not None):
908
flask.abort(403)
909
app.logger.info(f"Loading {server_repo_location}")
911
if not os.path.exists(server_repo_location):
913
app.logger.error(f"Cannot load {server_repo_location}")
914
return flask.render_template("not-found.html"), 404
915
repo = git.Repo(server_repo_location)
917
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
918
user = User.query.filter_by(username=flask.session.get("username")).first()
919
relationships = RepoAccess.query.filter_by(repo=repo_data)
920
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
921
if not user:
922
flask.abort(401)
923
post = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
925
if score:
927
old_relationship = PostVote.query.filter_by(user_username=user.username,
928
post_identifier=post.identifier).first()
929
if old_relationship:
930
if score == old_relationship.vote_score:
931
db.session.delete(old_relationship)
932
post.vote_sum -= old_relationship.vote_score
933
else:
934
post.vote_sum -= old_relationship.vote_score
935
post.vote_sum += score
936
old_relationship.vote_score = score
937
else:
938
relationship = PostVote(user, post, score)
939
post.vote_sum += score
940
db.session.add(relationship)
941
db.session.commit()
943
user_vote = PostVote.query.filter_by(user_username=user.username,
945
post_identifier=post.identifier).first()
946
response = flask.make_response(
947
str(post.vote_sum) + " " + str(user_vote.vote_score if user_vote else 0))
948
response.content_type = "text/plain"
949
return response
951
@repositories.route("/<username>/<repository>/favourite")
954
def repository_favourite(username, repository):
955
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
956
if not os.path.exists(server_repo_location):
957
app.logger.error(f"Cannot load {server_repo_location}")
958
flask.abort(404)
959
if not (get_visibility(username, repository) or get_permission_level(
960
flask.session.get("username"), username,
961
repository) is not None):
962
flask.abort(403)
963
app.logger.info(f"Loading {server_repo_location}")
965
if not os.path.exists(server_repo_location):
967
app.logger.error(f"Cannot load {server_repo_location}")
968
return flask.render_template("not-found.html"), 404
969
repo = git.Repo(server_repo_location)
971
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
972
user = User.query.filter_by(username=flask.session.get("username")).first()
973
relationships = RepoAccess.query.filter_by(repo=repo_data)
974
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
975
if not user:
976
flask.abort(401)
977
old_relationship = RepoFavourite.query.filter_by(user_username=user.username,
979
repo_route=repo_data.route).first()
980
if old_relationship:
981
db.session.delete(old_relationship)
982
else:
983
relationship = RepoFavourite(user, repo_data)
984
db.session.add(relationship)
985
db.session.commit()
987
return flask.redirect(flask.url_for("favourites"), code=303)
989
@repositories.route("/<username>/<repository>/users/", methods=["GET", "POST"])
992
def repository_users(username, repository):
993
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
994
if not os.path.exists(server_repo_location):
995
app.logger.error(f"Cannot load {server_repo_location}")
996
flask.abort(404)
997
if not (get_visibility(username, repository) or get_permission_level(
998
flask.session.get("username"), username,
999
repository) is not None):
1000
flask.abort(403)
1001
app.logger.info(f"Loading {server_repo_location}")
1003
if not os.path.exists(server_repo_location):
1005
app.logger.error(f"Cannot load {server_repo_location}")
1006
return flask.render_template("not-found.html"), 404
1007
repo = git.Repo(server_repo_location)
1009
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1010
user = User.query.filter_by(username=flask.session.get("username")).first()
1011
relationships = RepoAccess.query.filter_by(repo=repo_data)
1012
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
1013
if flask.request.method == "GET":
1015
return flask.render_template(
1016
"repo-users.html",
1017
username=username,
1018
repository=repository,
1019
repo_data=repo_data,
1020
relationships=relationships,
1021
repo=repo,
1022
user_relationship=user_relationship,
1023
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1024
is_favourite=get_favourite(flask.session.get("username"), username, repository)
1025
)
1026
else:
1027
if get_permission_level(flask.session.get("username"), username, repository) != 2:
1028
flask.abort(401)
1029
if flask.request.form.get("new-username"):
1031
# Create new relationship
1032
new_user = User.query.filter_by(
1033
username=flask.request.form.get("new-username")).first()
1034
relationship = RepoAccess(new_user, repo_data, flask.request.form.get("new-level"))
1035
db.session.add(relationship)
1036
db.session.commit()
1037
if flask.request.form.get("update-username"):
1038
# Create new relationship
1039
updated_user = User.query.filter_by(
1040
username=flask.request.form.get("update-username")).first()
1041
relationship = RepoAccess.query.filter_by(repo=repo_data, user=updated_user).first()
1042
if flask.request.form.get("update-level") == -1:
1043
relationship.delete()
1044
else:
1045
relationship.access_level = flask.request.form.get("update-level")
1046
db.session.commit()
1047
return flask.redirect(
1049
app.url_for(".repository_users", username=username, repository=repository))
1050
@repositories.route("/<username>/<repository>/branches/")
1053
def repository_branches(username, repository):
1054
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1055
if not os.path.exists(server_repo_location):
1056
app.logger.error(f"Cannot load {server_repo_location}")
1057
flask.abort(404)
1058
if not (get_visibility(username, repository) or get_permission_level(
1059
flask.session.get("username"), username,
1060
repository) is not None):
1061
flask.abort(403)
1062
app.logger.info(f"Loading {server_repo_location}")
1064
if not os.path.exists(server_repo_location):
1066
app.logger.error(f"Cannot load {server_repo_location}")
1067
return flask.render_template("not-found.html"), 404
1068
repo = git.Repo(server_repo_location)
1070
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1071
return flask.render_template(
1073
"repo-branches.html",
1074
username=username,
1075
repository=repository,
1076
repo_data=repo_data,
1077
repo=repo,
1078
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1079
is_favourite=get_favourite(flask.session.get("username"), username, repository)
1080
)
1081
@repositories.route("/<username>/<repository>/log/", defaults={"branch": None})
1084
@repositories.route("/<username>/<repository>/log/<branch>/")
1085
def repository_log(username, repository, branch):
1086
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1087
if not os.path.exists(server_repo_location):
1088
app.logger.error(f"Cannot load {server_repo_location}")
1089
flask.abort(404)
1090
if not (get_visibility(username, repository) or get_permission_level(
1091
flask.session.get("username"), username,
1092
repository) is not None):
1093
flask.abort(403)
1094
app.logger.info(f"Loading {server_repo_location}")
1096
if not os.path.exists(server_repo_location):
1098
app.logger.error(f"Cannot load {server_repo_location}")
1099
return flask.render_template("not-found.html"), 404
1100
repo = git.Repo(server_repo_location)
1102
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1103
if not repo_data.default_branch:
1104
if repo.heads:
1105
repo_data.default_branch = repo.heads[0].name
1106
else:
1107
return flask.render_template("empty.html",
1108
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
1109
if not branch:
1110
branch = repo_data.default_branch
1111
return flask.redirect(f"./{branch}", code=302)
1112
if branch.startswith("tag:"):
1114
ref = f"tags/{branch[4:]}"
1115
elif branch.startswith("~"):
1116
ref = branch[1:]
1117
else:
1118
ref = f"heads/{branch}"
1119
ref = ref.replace("~", "/") # encode slashes for URL support
1121
try:
1123
repo.git.checkout("-f", ref)
1124
except git.exc.GitCommandError:
1125
return flask.render_template("not-found.html"), 404
1126
branches = repo.heads
1128
all_refs = []
1130
for ref in repo.heads:
1131
all_refs.append((ref, "head"))
1132
for ref in repo.tags:
1133
all_refs.append((ref, "tag"))
1134
commit_list = [f"/{username}/{repository}/{sha}" for sha in
1136
git_command(server_repo_location, None, "log",
1137
"--format='%H'").decode().split("\n")]
1138
commits = Commit.query.filter(Commit.identifier.in_(commit_list))
1140
return flask.render_template(
1142
"repo-log.html",
1143
username=username,
1144
repository=repository,
1145
branches=all_refs,
1146
current=branch,
1147
repo_data=repo_data,
1148
repo=repo,
1149
commits=commits,
1150
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1151
is_favourite=get_favourite(flask.session.get("username"), username, repository)
1152
)
1153
@repositories.route("/<username>/<repository>/prs/", methods=["GET", "POST"])
1156
def repository_prs(username, repository):
1157
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1158
if not os.path.exists(server_repo_location):
1159
app.logger.error(f"Cannot load {server_repo_location}")
1160
flask.abort(404)
1161
if not (get_visibility(username, repository) or get_permission_level(
1162
flask.session.get("username"), username,
1163
repository) is not None):
1164
flask.abort(403)
1165
app.logger.info(f"Loading {server_repo_location}")
1167
if not os.path.exists(server_repo_location):
1169
app.logger.error(f"Cannot load {server_repo_location}")
1170
return flask.render_template("not-found.html"), 404
1171
if flask.request.method == "GET":
1173
repo = git.Repo(server_repo_location)
1174
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1175
user = User.query.filter_by(username=flask.session.get("username")).first()
1176
return flask.render_template(
1178
"repo-prs.html",
1179
username=username,
1180
repository=repository,
1181
repo_data=repo_data,
1182
repo=repo,
1183
PullRequest=PullRequest,
1184
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1185
is_favourite=get_favourite(flask.session.get("username"), username, repository),
1186
default_branch=repo_data.default_branch,
1187
branches=repo.branches
1188
)
1189
else:
1191
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1192
head = flask.request.form.get("head")
1193
head_route = flask.request.form.get("headroute")
1194
base = flask.request.form.get("base")
1195
if not head and base and head_route:
1197
return flask.redirect(".", 400)
1198
head_repo = git.Repo(os.path.join(config.REPOS_PATH, head_route.lstrip("/")))
1200
base_repo = git.Repo(server_repo_location)
1201
print(head_repo)
1202
if head not in head_repo.branches or base not in base_repo.branches:
1204
flask.flash(Markup(
1205
"<iconify-icon icon='mdi:error'></iconify-icon>" + _("Bad branch name")),
1206
category="error")
1207
return flask.redirect(".", 303)
1208
head_data = db.session.get(Repo, head_route)
1210
if not head_data.visibility:
1211
flask.flash(Markup(
1212
"<iconify-icon icon='mdi:error'></iconify-icon>" + _(
1213
"Head can't be restricted")),
1214
category="error")
1215
return flask.redirect(".", 303)
1216
pull_request = PullRequest(repo_data, head, head_data, base,
1218
db.session.get(User, flask.session["username"]))
1219
db.session.add(pull_request)
1221
db.session.commit()
1222
return flask.redirect(".", 303)
1224
@repositories.route("/<username>/<repository>/prs/merge", methods=["POST"])
1227
def repository_prs_merge(username, repository):
1228
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1229
if not os.path.exists(server_repo_location):
1230
app.logger.error(f"Cannot load {server_repo_location}")
1231
flask.abort(404)
1232
if not (get_visibility(username, repository) or get_permission_level(
1233
flask.session.get("username"), username,
1234
repository) is not None):
1235
flask.abort(403)
1236
if not get_permission_level(flask.session.get("username"), username, repository):
1238
flask.abort(401)
1239
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1241
repo = git.Repo(server_repo_location)
1242
id = flask.request.form.get("id")
1243
pull_request = db.session.get(PullRequest, id)
1245
if pull_request:
1247
result = celery_tasks.merge_heads.delay(
1248
pull_request.head_route,
1249
pull_request.head_branch,
1250
pull_request.base_route,
1251
pull_request.base_branch,
1252
simulate=True
1253
)
1254
task_result = worker.AsyncResult(result.id)
1255
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303)
1257
# db.session.delete(pull_request)
1258
# db.session.commit()
1259
else:
1260
flask.abort(400)
1261
@repositories.route("/<username>/<repository>/prs/<int:id>/merge")
1264
def repository_prs_merge_stage_two(username, repository, id):
1265
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1266
if not os.path.exists(server_repo_location):
1267
app.logger.error(f"Cannot load {server_repo_location}")
1268
flask.abort(404)
1269
if not (get_visibility(username, repository) or get_permission_level(
1270
flask.session.get("username"), username,
1271
repository) is not None):
1272
flask.abort(403)
1273
if not get_permission_level(flask.session.get("username"), username, repository):
1275
flask.abort(401)
1276
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1278
repo = git.Repo(server_repo_location)
1279
pull_request = db.session.get(PullRequest, id)
1281
if pull_request:
1283
result = celery_tasks.merge_heads.delay(
1284
pull_request.head_route,
1285
pull_request.head_branch,
1286
pull_request.base_route,
1287
pull_request.base_branch,
1288
simulate=False
1289
)
1290
task_result = worker.AsyncResult(result.id)
1291
pull_request.state = 1
1293
db.session.commit()
1294
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303)
1296
# db.session.delete(pull_request)
1297
else:
1298
flask.abort(400)
1299
@app.route("/task/<task_id>")
1302
def task_monitor(task_id):
1303
task_result = worker.AsyncResult(task_id)
1304
print(task_result.status)
1305
return flask.render_template("task-monitor.html", result=task_result)
1307
@repositories.route("/<username>/<repository>/prs/delete", methods=["POST"])
1310
def repository_prs_delete(username, repository):
1311
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1312
if not os.path.exists(server_repo_location):
1313
app.logger.error(f"Cannot load {server_repo_location}")
1314
flask.abort(404)
1315
if not (get_visibility(username, repository) or get_permission_level(
1316
flask.session.get("username"), username,
1317
repository) is not None):
1318
flask.abort(403)
1319
if not get_permission_level(flask.session.get("username"), username, repository):
1321
flask.abort(401)
1322
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1324
repo = git.Repo(server_repo_location)
1325
id = flask.request.form.get("id")
1326
pull_request = db.session.get(PullRequest, id)
1328
if pull_request:
1330
pull_request.state = 2
1331
db.session.commit()
1332
return flask.redirect(".", 303)
1334
@repositories.route("/<username>/<repository>/settings/")
1337
def repository_settings(username, repository):
1338
if get_permission_level(flask.session.get("username"), username, repository) != 2:
1339
flask.abort(401)
1340
return flask.render_template("repo-settings.html", username=username, repository=repository)
1342
@app.errorhandler(404)
1345
def e404(error):
1346
return flask.render_template("not-found.html"), 404
1347
@app.errorhandler(401)
1350
def e401(error):
1351
return flask.render_template("unauthorised.html"), 401
1352
@app.errorhandler(403)
1355
def e403(error):
1356
return flask.render_template("forbidden.html"), 403
1357
@app.errorhandler(418)
1360
def e418(error):
1361
return flask.render_template("teapot.html"), 418
1362
@app.errorhandler(405)
1365
def e405(error):
1366
return flask.render_template("method-not-allowed.html"), 405
1367
if __name__ == "__main__":
1370
app.run(debug=True, port=8080, host="0.0.0.0")
1371
app.register_blueprint(repositories)