import os import random import subprocess from functools import wraps import cairosvg import flask from flask_sqlalchemy import SQLAlchemy import git import mimetypes import magic from flask_bcrypt import Bcrypt from markupsafe import escape, Markup from flask_migrate import Migrate from datetime import datetime from enum import Enum import shutil from PIL import Image from cairosvg import svg2png import platform import config app = flask.Flask(__name__) from flask_httpauth import HTTPBasicAuth auth = HTTPBasicAuth() app.config["SQLALCHEMY_DATABASE_URI"] = config.DB_URI app.config["SECRET_KEY"] = config.DB_PASSWORD app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False db = SQLAlchemy(app) bcrypt = Bcrypt(app) migrate = Migrate(app, db) def gitCommand(repo, data, *args): if not os.path.isdir(repo): raise FileNotFoundError("Repo not found") env = os.environ.copy() command = ["git", *args] proc = subprocess.Popen(" ".join(command), cwd=repo, env=env, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE) print(command) if data: proc.stdin.write(data) out, err = proc.communicate() return out def onlyChars(string, chars): for i in string: if i not in chars: return False return True with app.app_context(): class RepoAccess(db.Model): id = db.Column(db.Integer, primary_key=True) userUsername = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) repoRoute = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) accessLevel = db.Column(db.SmallInteger(), nullable=False) # 0 read-only, 1 read-write, 2 admin user = db.relationship("User", back_populates="repoAccess") repo = db.relationship("Repo", back_populates="repoAccess") __table_args__ = (db.UniqueConstraint("userUsername", "repoRoute", name="_user_repo_uc"),) def __init__(self, user, repo, level): self.userUsername = user.username self.repoRoute = repo.route self.accessLevel = level class User(db.Model): username = db.Column(db.String(32), unique=True, nullable=False, primary_key=True) displayName = db.Column(db.Unicode(128), unique=False, nullable=True) bio = db.Column(db.Unicode(512), unique=False, nullable=True) passwordHashed = db.Column(db.String(60), nullable=False) email = db.Column(db.String(254), nullable=True) company = db.Column(db.Unicode(64), nullable=True) companyURL = db.Column(db.String(256), nullable=True) URL = db.Column(db.String(256), nullable=True) showMail = db.Column(db.Boolean, default=False, nullable=False) location = db.Column(db.Unicode(64), nullable=True) creationDate = db.Column(db.DateTime, default=datetime.utcnow) repositories = db.relationship("Repo", back_populates="owner") repoAccess = db.relationship("RepoAccess", back_populates="user") def __init__(self, username, password, email=None, displayName=None): self.username = username self.passwordHashed = bcrypt.generate_password_hash(password, config.HASHING_ROUNDS).decode("utf-8") self.email = email self.displayName = displayName # Create the user's directory if not os.path.exists(os.path.join(config.REPOS_PATH, username)): os.makedirs(os.path.join(config.REPOS_PATH, username)) if not os.path.exists(os.path.join(config.USERDATA_PATH, username)): os.makedirs(os.path.join(config.USERDATA_PATH, username)) avatarName = random.choice(os.listdir(config.DEFAULT_AVATARS_PATH)) if os.path.join(config.DEFAULT_AVATARS_PATH, avatarName).endswith(".svg"): cairosvg.svg2png(url=os.path.join(config.DEFAULT_AVATARS_PATH, avatarName), write_to="/tmp/roundabout-avatar.png") avatar = Image.open("/tmp/roundabout-avatar.png") else: avatar = Image.open(os.path.join(config.DEFAULT_AVATARS_PATH, avatarName)) avatar.thumbnail(config.AVATAR_SIZE) avatar.save(os.path.join(config.USERDATA_PATH, username, "avatar.png")) class Repo(db.Model): route = db.Column(db.String(98), unique=True, nullable=False, primary_key=True) ownerName = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) name = db.Column(db.String(64), nullable=False) owner = db.relationship("User", back_populates="repositories") visibility = db.Column(db.SmallInteger(), nullable=False) info = db.Column(db.Unicode(512), nullable=True) URL = db.Column(db.String(256), nullable=True) creationDate = db.Column(db.DateTime, default=datetime.utcnow) defaultBranch = db.Column(db.String(64), nullable=True, default="") commits = db.relationship("Commit", back_populates="repo") repoAccess = db.relationship("RepoAccess", back_populates="repo") def __init__(self, owner, name, visibility): self.route = f"/{owner.username}/{name}" self.name = name self.ownerName = owner.username self.owner = owner self.visibility = visibility # Add the owner as an admin repoAccess = RepoAccess(owner, self, 2) db.session.add(repoAccess) class Commit(db.Model): identifier = db.Column(db.String(227), unique=True, nullable=False, primary_key=True) sha = db.Column(db.String(128), nullable=False) repoName = db.Column(db.String(97), db.ForeignKey("repo.route"), nullable=False) ownerName = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) ownerIdentity = db.Column(db.String(321)) receiveDate = db.Column(db.DateTime, default=datetime.now) authorDate = db.Column(db.DateTime) message = db.Column(db.UnicodeText) repo = db.relationship("Repo", back_populates="commits") def __init__(self, sha, owner, repo, date, message, ownerIdentity): self.identifier = f"/{owner.username}/{repo.name}/{sha}" self.sha = sha self.repoName = repo.route self.repo = repo self.ownerName = owner.username self.owner = owner self.authorDate = datetime.fromtimestamp(int(date)) self.message = message self.ownerIdentity = ownerIdentity def getPermissionLevel(loggedIn, username, repository): user = User.query.filter_by(username=loggedIn).first() repo = Repo.query.filter_by(route=f"/{username}/{repository}").first() if user and repo: permission = RepoAccess.query.filter_by(user=user, repo=repo).first() if permission: return permission.accessLevel return None def getVisibility(username, repository): repo = Repo.query.filter_by(route=f"/{username}/{repository}").first() if repo: return repo.visibility return None import gitHTTP import jinjaUtils def humanSize(value, decimals=2, scale=1024, units=("B", "kiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB", "RiB", "QiB")): for unit in units: if value < scale: break value /= scale if int(value) == value: # do not return decimals, if the value is already round return int(value), unit return round(value * 10 ** decimals) / 10 ** decimals, unit def guessMIME(path): if os.path.isdir(path): mimetype = "inode/directory" elif magic.from_file(path, mime=True): mimetype = magic.from_file(path, mime=True) else: mimetype = "application/octet-stream" return mimetype def convertToHTML(path): with open(path, "r") as f: contents = f.read() return contents @app.context_processor def default(): username = flask.session.get("username") return {"loggedInUser": username} @app.route("/") def main(): return flask.render_template("home.html") @app.route("/about/") def about(): return flask.render_template("about.html", platform=platform) @app.route("/settings/", methods=["GET", "POST"]) def settings(): if flask.request.method == "GET": if not flask.session.get("username"): flask.abort(401) user = User.query.filter_by(username=flask.session.get("username")).first() return flask.render_template("user-settings.html", user=user) else: user = User.query.filter_by(username=flask.session.get("username")).first() user.displayName = flask.request.form["displayname"] user.URL = flask.request.form["url"] user.company = flask.request.form["company"] user.companyURL = flask.request.form["companyurl"] user.location = flask.request.form["location"] user.showMail = flask.request.form.get("showmail", user.showMail) db.session.commit() flask.flash(Markup("Settings saved"), category="success") return flask.redirect(f"/{flask.session.get('username')}", code=303) @app.route("/accounts/", methods=["GET", "POST"]) def login(): if flask.request.method == "GET": return flask.render_template("login.html") else: if "login" in flask.request.form: username = flask.request.form["username"] password = flask.request.form["password"] user = User.query.filter_by(username=username).first() if user and bcrypt.check_password_hash(user.passwordHashed, password): flask.session["username"] = user.username flask.flash( Markup(f"Successfully logged in as {username}"), category="success") return flask.redirect("/", code=303) elif not user: flask.flash(Markup("User not found"), category="alert") return flask.render_template("login.html") else: flask.flash(Markup("Invalid password"), category="error") return flask.render_template("login.html") if "signup" in flask.request.form: username = flask.request.form["username"] password = flask.request.form["password"] password2 = flask.request.form["password2"] email = flask.request.form.get("email") email2 = flask.request.form.get("email2") # repeat email is a honeypot name = flask.request.form.get("name") if not onlyChars(username, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"): flask.flash(Markup( "Usernames may only contain Latin alphabet, numbers, '-' and '_'"), category="error") return flask.render_template("login.html") if username in config.RESERVED_NAMES: flask.flash( Markup(f"Sorry, {username} is a system path"), category="error") return flask.render_template("login.html") userCheck = User.query.filter_by(username=username).first() if userCheck: flask.flash( Markup(f"The username {username} is taken"), category="error") return flask.render_template("login.html") if password2 != password: flask.flash(Markup("Make sure the passwords match"), category="error") return flask.render_template("login.html") user = User(username, password, email, name) db.session.add(user) db.session.commit() flask.session["username"] = user.username flask.flash(Markup( f"Successfully created and logged in as {username}"), category="success") return flask.redirect("/", code=303) @app.route("/newrepo/", methods=["GET", "POST"]) def newRepo(): if flask.request.method == "GET": return flask.render_template("new-repo.html") else: name = flask.request.form["name"] visibility = int(flask.request.form["visibility"]) if not onlyChars(name, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"): flask.flash(Markup( "Repository names may only contain Latin alphabet, numbers, '-' and '_'"), category="error") return flask.render_template("new-repo.html") user = User.query.filter_by(username=flask.session.get("username")).first() repo = Repo(user, name, visibility) db.session.add(repo) db.session.commit() if not os.path.exists(os.path.join(config.REPOS_PATH, repo.route)): subprocess.run(["git", "init", repo.name], cwd=os.path.join(config.REPOS_PATH, flask.session.get("username"))) flask.flash(Markup(f"Successfully created repository {name}"), category="success") return flask.redirect(repo.route, code=303) @app.route("/logout") def logout(): flask.session.clear() flask.flash(Markup(f"Successfully logged out"), category="info") return flask.redirect("/", code=303) @app.route("//") def userProfile(username): user = User.query.filter_by(username=username).first() repos = Repo.query.filter_by(ownerName=username, visibility=2) return flask.render_template("user-profile.html", user=user, repos=repos) @app.route("///") def repositoryIndex(username, repository): return flask.redirect("./tree", code=302) @app.route("///raw//") def repositoryRaw(username, repository, branch, subpath): if not (getVisibility(username, repository) or getPermissionLevel(flask.session.get("user"), username, repository) is not None): flask.abort(403) serverRepoLocation = os.path.join(config.REPOS_PATH, os.path.join(username, repository)) app.logger.info(f"Loading {serverRepoLocation}") if not os.path.exists(serverRepoLocation): app.logger.error(f"Cannot load {serverRepoLocation}") return flask.render_template("not-found.html"), 404 repo = git.Repo(serverRepoLocation) repoData = Repo.query.filter_by(route=f"/{username}/{repository}").first() if not repoData.defaultBranch: if repo.heads: repoData.defaultBranch = repo.heads[0].name else: return flask.render_template("empty.html", remote=f"http://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200 if not branch: branch = repoData.defaultBranch return flask.redirect(f"./{branch}", code=302) if branch.startswith("tag:"): ref = f"tags/{branch[4:]}" elif branch.startswith("~"): ref = branch[1:] else: ref = f"heads/{branch}" ref = ref.replace("~", "/") # encode slashes for URL support try: repo.git.checkout("-f", ref) except git.exc.GitCommandError: return flask.render_template("not-found.html"), 404 return flask.send_from_directory(config.REPOS_PATH, os.path.join(username, repository, subpath)) @app.route("/info//avatar") def userAvatar(username): serverUserdataLocation = os.path.join(config.USERDATA_PATH, username) if not os.path.exists(serverUserdataLocation): return flask.render_template("not-found.html"), 404 return flask.send_from_directory(serverUserdataLocation, "avatar.png") @app.route("///tree/", defaults={"branch": None, "subpath": ""}) @app.route("///tree//", defaults={"subpath": ""}) @app.route("///tree//") def repositoryTree(username, repository, branch, subpath): if not (getVisibility(username, repository) or getPermissionLevel(flask.session.get("username"), username, repository) is not None): flask.abort(403) serverRepoLocation = os.path.join(config.REPOS_PATH, os.path.join(username, repository)) app.logger.info(f"Loading {serverRepoLocation}") if not os.path.exists(serverRepoLocation): app.logger.error(f"Cannot load {serverRepoLocation}") return flask.render_template("not-found.html"), 404 repo = git.Repo(serverRepoLocation) repoData = Repo.query.filter_by(route=f"/{username}/{repository}").first() if not repoData.defaultBranch: if repo.heads: repoData.defaultBranch = repo.heads[0].name else: return flask.render_template("empty.html", remote=f"http://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200 if not branch: branch = repoData.defaultBranch return flask.redirect(f"./{branch}", code=302) if branch.startswith("tag:"): ref = f"tags/{branch[4:]}" elif branch.startswith("~"): ref = branch[1:] else: ref = f"heads/{branch}" ref = ref.replace("~", "/") # encode slashes for URL support try: repo.git.checkout("-f", ref) except git.exc.GitCommandError: return flask.render_template("not-found.html"), 404 branches = repo.heads allRefs = [] for ref in repo.heads: allRefs.append((ref, "head")) for ref in repo.tags: allRefs.append((ref, "tag")) if os.path.isdir(os.path.join(serverRepoLocation, subpath)): files = [] blobs = [] for entry in os.listdir(os.path.join(serverRepoLocation, subpath)): if not os.path.basename(entry) == ".git": files.append(os.path.join(subpath, entry)) infos = [] for file in files: path = os.path.join(serverRepoLocation, file) mimetype = guessMIME(path) text = gitCommand(serverRepoLocation, None, "log", "--format='%H\n'", file).decode() sha = text.split("\n")[0] identifier = f"/{username}/{repository}/{sha}" lastCommit = Commit.query.filter_by(identifier=identifier).first() info = { "name": os.path.basename(file), "serverPath": path, "relativePath": file, "link": os.path.join(f"/{username}/{repository}/tree/{branch}/", file), "size": humanSize(os.path.getsize(path)), "mimetype": f"{mimetype}{f' ({mimetypes.guess_type(path)[1]})' if mimetypes.guess_type(path)[1] else ''}", "commit": lastCommit, "shaSize": 7, } specialIcon = config.matchIcon(os.path.basename(file)) if specialIcon: info["icon"] = specialIcon elif os.path.isdir(path): info["icon"] = config.folderIcon elif mimetypes.guess_type(path)[0] in config.fileIcons: info["icon"] = config.fileIcons[mimetypes.guess_type(path)[0]] else: info["icon"] = config.unknownIcon if os.path.isdir(path): infos.insert(0, info) else: infos.append(info) return flask.render_template( "repo-tree.html", username=username, repository=repository, files=infos, subpath=os.path.join("/", subpath), branches=allRefs, current=branch ) else: path = os.path.join(serverRepoLocation, subpath) if not os.path.exists(path): return flask.render_template("not-found.html"), 404 mimetype = guessMIME(path) mode = mimetype.split("/", 1)[0] size = humanSize(os.path.getsize(path)) specialIcon = config.matchIcon(os.path.basename(path)) if specialIcon: icon = specialIcon elif os.path.isdir(path): icon = config.folderIcon elif mimetypes.guess_type(path)[0] in config.fileIcons: icon = config.fileIcons[mimetypes.guess_type(path)[0]] else: icon = config.unknownIcon contents = None if mode == "text": contents = convertToHTML(path) return flask.render_template( "repo-file.html", username=username, repository=repository, file=os.path.join(f"/{username}/{repository}/raw/{branch}/", subpath), branches=allRefs, current=branch, mode=mode, mimetype=mimetype, detailedtype=magic.from_file(path), size=size, icon=icon, subpath=os.path.join("/", subpath), basename=os.path.basename(path), contents=contents ) @app.route("///forum/") def repositoryForum(username, repository): if not (getVisibility(username, repository) or getPermissionLevel(flask.session.get("username"), username, repository) is not None): flask.abort(403) return flask.render_template("repo-forum.html", username=username, repository=repository) @app.route("///branches/") def repositoryBranches(username, repository): if not (getVisibility(username, repository) or getPermissionLevel(flask.session.get("username"), username, repository) is not None): flask.abort(403) serverRepoLocation = os.path.join(config.REPOS_PATH, os.path.join(username, repository)) app.logger.info(f"Loading {serverRepoLocation}") if not os.path.exists(serverRepoLocation): app.logger.error(f"Cannot load {serverRepoLocation}") return flask.render_template("not-found.html"), 404 repo = git.Repo(serverRepoLocation) repoData = Repo.query.filter_by(route=f"/{username}/{repository}").first() return flask.render_template("repo-branches.html", username=username, repository=repository, repoData=repoData, repo=repo) @app.route("///log/") def repositoryLog(username, repository): if not (getVisibility(username, repository) or getPermissionLevel(flask.session.get("username"), username, repository) is not None): flask.abort(403) serverRepoLocation = os.path.join(config.REPOS_PATH, os.path.join(username, repository)) app.logger.info(f"Loading {serverRepoLocation}") if not os.path.exists(serverRepoLocation): app.logger.error(f"Cannot load {serverRepoLocation}") return flask.render_template("not-found.html"), 404 repo = git.Repo(serverRepoLocation) repoData = Repo.query.filter_by(route=f"/{username}/{repository}").first() commits = Commit.query.filter_by(repo=repoData) return flask.render_template("repo-log.html", username=username, repository=repository, repoData=repoData, repo=repo, commits=commits) @app.route("///settings/") def repositorySettings(username, repository): if getPermissionLevel(flask.session.get("username"), username, repository) != 2: flask.abort(401) return flask.render_template("repo-settings.html", username=username, repository=repository) @app.errorhandler(404) def e404(error): return flask.render_template("not-found.html"), 404 @app.errorhandler(401) def e401(error): return flask.render_template("unauthorised.html"), 401 @app.errorhandler(403) def e403(error): return flask.render_template("forbidden.html"), 403 @app.errorhandler(418) def e418(error): return flask.render_template("teapot.html"), 418 if __name__ == "__main__": app.run(debug=True, port=8080, host="0.0.0.0")