main.py

#!/usr/bin/env python3
from waitress import serve #Used as webserver (Production)
from flask import Flask, request, render_template, redirect, abort, Markup, session, make_response #Used to prepare the dynamic pages (The main site)
import sqlite3 #Used to store the Data
import os #Used for getting the enviorement variables
import qrcode #Used to generate the QR
import base64 #Used to encode the generated QR as base64, to directly insert it into the HTML
from requests import post, get #Used to validate recaptcha / oauth
from io import BytesIO #Needed for base64 encoding of the image
from PIL import Image #Needed for QR generation
from flask_github import GitHub #github oauth library
import json #used for github oauth
from html import escape #This is used to escape characters, if they are send in the url
app = Flask(__name__)

domain_to_index = {}

try:
    domain = os.environ["domains"].split(";") #Get the domains from the enviorement variable. If no enviorement variable is set set it to 127.0.0.1:5000 (for testing)
except:
    domain = ["127.0.0.1:5000"]

try:
    if(os.environ["show_build_date"] == "1"): #If you want to see the builddate you can enable this enviorement variable
        builddate = open("builddate.txt", "r").read()
    else:
        builddate = ""
except:
    builddate = "" #If the enviorement Variable is not set also skip the builddate

try:
    if(os.environ["show_version"] == "1"): #If you want to see the builddate you can enable this enviorement variable
        version = open("VERSION", "r").read()
    else:
        version = ""
except:
    version = "" #If the enviorement Variable is not set also skip the version

try:
    recaptchaPrivateKey = os.environ["recaptcha_private"] #Get the recaptcha keys, if not set set skipRecaptcha to true to skip the check. If the publicKey is not set the user also will not get the code needed for recaptcha delivered in the page.
    recaptchaPublicKey = os.environ["recaptcha_public"]
    if(recaptchaPrivateKey != "") and (recaptchaPublicKey != ""): #If the variables are empty also skip the captcha
        skipCaptcha = False
    else:
        skipCaptcha = True

except:
    recaptchaPrivateKey = ""
    recaptchaPublicKey = ""
    skipCaptcha = True

try:
    if(os.environ["production"] == "1"): #If you use this in production, please set this to 1, because the Flask Testserver is not very secure (e.g. shows error on Website)
        production = True
    else:
        production = False
except:
    production = False

try: #If you use https with a proxy afterwards you can set this to https and internal redirects will be https not http. This is to prevent bugs with modern browsers, bacause they block http content if the main page is loaded via https.
    url_scheme = os.environ["url_scheme"]
except:
    url_scheme = "http"

try:
    host=os.environ["host"]
except:
    host="127.0.0.1"

try:
    app.config['GITHUB_CLIENT_ID'] = os.environ['GITHUB_CLIENT_ID']
    app.config['GITHUB_CLIENT_SECRET'] = os.environ['GITHUB_CLIENT_SECRET']
except:
    print("github client id or client secret is not set, please set these and run again.")
    exit()
github = GitHub(app)

index = 0
domain_prepared = ""
for domains in domain: #Make from every domnain a entry for the select box later
    domain_prepared = domain_prepared + '<option value="' + str(domains) + '">' + str(domains) + '</option>'
    domain_to_index[domains] = str(index)
    index = index + 1
if(index > 1):
    showDomainSelect=True #Show only domain select, if there are more than one available
else:
    showDomainSelect=False
    domain_prepared = domain[0]


def table_check(): #This function is used on start to make a new Database if not already exists.
    create_table = """
        CREATE TABLE WEB_URL(
        LONG_URL TEXT NOT NULL,
        SHORT_URL TEXT NOT NULL,
        USERNAME TEXT
        );
        """
    with sqlite3.connect('db/urls.db') as conn:
        cursor = conn.cursor()
        try: #Try making the database structure, if fails Database was already created.
            cursor.execute(create_table)
        except sqlite3.OperationalError:
            pass


def makeQR(text): #This function is used to create a QR code and encode it base64, if you make a new shortlink
    qr = qrcode.QRCode( #QR generation variables
        version=1,
        error_correction=qrcode.constants.ERROR_CORRECT_L,
        box_size=10,
        border=1,
    )
    qr.add_data(text) #The URL is in the text variable
    qr.make(fit=True) #Generate the QR

    img = qr.make_image(fill_color="black", back_color="white") #Encode the WR as base 64
    with BytesIO() as buffer:
        img.save(buffer, 'jpeg')
        return base64.b64encode(buffer.getvalue()).decode()


def grecaptcha_verify(request): #This function is used to verify the google recaptcha code, that is send to the server after submitting a new link
    captcha_rs = request.form.get('g-recaptcha-response')
    url = "https://www.google.com/recaptcha/api/siteverify" #The baseurl
    headers = {'User-Agent': 'DebuguearApi-Browser',} #Useragent doesn't matters, but is set here
    params = {'secret': recaptchaPrivateKey, 'response': captcha_rs} #As paramtere we send to google our private Key and the key from the user
    verify_rs = post(url,params, headers=headers) #Send a post request with the parameters from before to googlde
    verify_rs = verify_rs.json()
    response = verify_rs.get("success", False) #Verify that the response includes the success, if so return True, if not return False
    return response 



@app.route('/', methods=['GET'])
def home_get():
    try:
        loginbar = "Hello " + request.cookies.get('username') + ' (<a href="/user/links" style="color:white">your links</a>, <a href="/user/logout" style="color:white">logout</a>)'
    except:
        loginbar = '<a href="/user/login" style="color:white">login</a>'

    return render_template('home.html', builddate=builddate, version=version, domain=domain_prepared, recaptchaPublicKey=recaptchaPublicKey, showDomainSelect=showDomainSelect, loginbar=loginbar) #return the default site to create a new shorten link


@app.route('/', methods=['POST']) #This function is used to create a new url
def home_post():
    
    try:
        userID = request.cookies.get('userID')
        loginbar = "Hello " + request.cookies.get('username') + ' (<a href="/user/links" style="color:white">your links</a>, <a href="/user/logout" style="color:white">logout</a>)'
    except:
        userID = "null"
        loginbar = '<a href="/user/login" style="color:white">login</a>'
    if not grecaptcha_verify(request) and not skipCaptcha:
        return render_template('home.html', builddate=builddate, version=version, domain=domain_prepared, snackbar="There was an error validating, that you are a human, please try again.", long_url_prefilled=request.form.get('url'), short_url_prefilled=request.form.get('short').lower(), domain_prefilled=domain_to_index[request.form.get('domain')], recaptchaPublicKey=recaptchaPublicKey, showDomainSelect=showDomainSelect, loginbar=loginbar) #return the user the prefilled form with an error message, because no url to short was provided
    
    if (request.form.get('url').replace(" ", "") == ""):
        return render_template('home.html', builddate=builddate, version=version, domain=domain_prepared, snackbar="Please enter a url to short, before submitting this form", long_url_prefilled=request.form.get('url'), short_url_prefilled=request.form.get('short').lower(), domain_prefilled=domain_to_index[request.form.get('domain')], recaptchaPublicKey=recaptchaPublicKey, showDomainSelect=showDomainSelect, loginbar=loginbar) #return the user the prefilled form with an error message, because no url to short was provided
    if (request.form.get('short').replace(" ", "") == ""):
        return render_template('home.html', builddate=builddate, version=version, domain=domain_prepared, snackbar="Please enter a short name, before submitting this form", long_url_prefilled=request.form.get('url'), short_url_prefilled=request.form.get('short').lower(), domain_prefilled=domain_to_index[request.form.get('domain')], recaptchaPublicKey=recaptchaPublicKey, showDomainSelect=showDomainSelect, loginbar=loginbar) #return the user the prefilled form with an error message, because no short link was provided
    shorturl = (request.form.get('domain') + "/" + request.form.get('short')).lower()

    url = request.form.get('url')
    with sqlite3.connect('db/urls.db') as conn: #Check if another user already used the short link
        cursor = conn.cursor()
        res = cursor.execute('SELECT LONG_URL FROM WEB_URL WHERE SHORT_URL=?', [shorturl])
        try:
            short = res.fetchone()
            already_used = False
            if short is not None:
                already_used = True
        except:
            pass

        if not already_used: #If short link wasn't used before, insert the link in the Database.
            res = cursor.execute(
                'INSERT INTO WEB_URL (LONG_URL, SHORT_URL, USERNAME) VALUES (?, ?, ?)',
                [url, shorturl, userID]
            )
            return render_template('home.html', short_url=shorturl, recaptchaPublicKey=recaptchaPublicKey, builddate=builddate, version=version, domain=domain_prepared, qrcode=makeQR("http://" + shorturl), loginbar=loginbar) #return the shorten link to the user
        else:
            return render_template('home.html', builddate=builddate, version=version, recaptchaPublicKey=recaptchaPublicKey, domain=domain_prepared, snackbar="URL already used, please try another one", long_url_prefilled=request.form.get('url'), short_url_prefilled=request.form.get('short').lower(), domain_prefilled=domain_to_index[request.form.get('domain')], showDomainSelect=showDomainSelect, loginbar=loginbar) #return the user the prefilled form with an error message, because the url was already used
    
@app.route('/favicon.ico') #Redirect to the static url of the favicon
def favicon():
    return redirect("/static/favicon.ico")

@app.route('/<short_url>')
def redirect_short_url(short_url):
    host = request.headers['Host']
    url = ""
    with sqlite3.connect('db/urls.db') as conn: #Get the original URL from the database
        cursor = conn.cursor()
        res = cursor.execute('SELECT LONG_URL FROM WEB_URL WHERE SHORT_URL=?', [host + "/" + short_url.lower()])
        try:
            short = res.fetchone()
            if short is not None: #If a long url is found
                url = short[0]
                error_404 = False
            else:
                error_404 = True #If no url is found throw a 404. If you throw a error in a try / catch block it will be catched by this, so set a variable to true and throw the error later
        except Exception as e: #If there happens an error, print the exception to the console and throw a 500 error
            print(e) #Print a debug Message to the console
            abort(500) #Throw a 500 error. This means internal Server error.
    if not error_404: #If there was no 404 error before, redirect the user. If not throw a 404 error
        return redirect(url)
    else:
        abort(404)


@app.route('/user/login')
def login():
    return github.authorize(scope="user") #redirect the user to the github login page and ask for access to user data (name, email, ...)

@app.route('/user/github-callback') #Github redirects to this link after the user authenticated. Then we use the Token we get from github and request via the github api the username and the userid
@github.authorized_handler
def authorized(oauth_token):
    if oauth_token is None:
        return "oauth failed, please try again" #If you call this page manual you get this error
    
    headers = {'Authorization': 'token ' + oauth_token,} #Useragent doesn't matters, but is set here
    githubResponse = get("https://api.github.com/user", headers=headers).text
    userID = str(json.loads(githubResponse)['id'])
    username = str(json.loads(githubResponse)['login'])

    resp = make_response(redirect('/')) #redirect the user at the end back to the main page
    resp.set_cookie('userID', userID) #set the cookies with username and userid
    resp.set_cookie('username', username)
    return resp
    
@app.route('/user/logout')
def logout():
    resp = make_response("logout successful")
    resp.set_cookie('userID', "", max_age=0) #Set the max age of the cookies to 0, this means delete the cookies.
    resp.set_cookie('username', "", max_age=0)
    return resp

@app.route('/user/links')#This function gives the user the posibility to see and delete his links
def ownLinks():
    try:
        userID = request.cookies.get('userID') #Get the userid from the cookie
        loginbar = "Hello " + request.cookies.get('username') + ' (<a href="/user/logout" style="color:white">logout</a>)' #This is the loginbar
    except:
        return redirect("/user/login") #If user is not logged in redirect him to the login page

    with sqlite3.connect('db/urls.db') as conn:
        cursor = conn.cursor()
        res = cursor.execute('SELECT LONG_URL, SHORT_URL FROM WEB_URL WHERE USERNAME=?', [userID]) #Get all entries from the database, that are created by this user
        response = '<table id="t01">\n<tr>\n<th>Long URL</th>\n<th>Short URL</th>\n<th>Action</th>\n</tr>\n' #This is the layout of the table
        try:
            entriesList = res.fetchall()
            for entries in entriesList: #for every entrie in the database add a line to the table
                response = response + "<tr>\n<td>" + entries[0] + "</td>\n<td>" + entries[1] + '</td>\n<td><a id="red" href="/user/delete?link=' + escape(entries[1]) + '">delete</a> <a href="#" id="dialog-link" onclick="buttonListener(\'' + entries[1] + '\', this)">QR</a></tr>\n'
            response = response + "</table>" #Close the table
            if(len(entriesList) == 0): response = 'you have no shorten links. <a href="/">back</a>' #If user has no shorten links make this message with a back button
        except:
            abort(500) #Shouldn't happen, 500 means internal server error
        return render_template('editEntries.html', content=response, loginbar=loginbar) #Put the table and the login div inside the template and server it to the user


@app.route('/user/delete') #This function is called if a user deletes an entrie
def delete():
    try:
        userID = request.cookies.get('userID') #get the userid from the cookie
        loginbar = "Hello " + request.cookies.get('username') + ' (<a href="/user/logout" style="color:white">logout</a>)' # generate the login form
    except:
        return redirect("/user/login") # if user is not logged in redirect him to the login page
    linkToDelete = request.args.get('link') #get the link, which the user want's to delete from the parameter in the url.

    with sqlite3.connect('db/urls.db') as conn:
        cursor = conn.cursor()
        try:
            cursor.execute('DELETE FROM WEB_URL WHERE SHORT_URL=? AND USERNAME=?', [linkToDelete, userID]) #Delete the entrie
            return redirect('/user/links') #redirect the user back to the table.
        except:
            abort(500)

@app.route('/user/makeqr')
def makeQrCode():
    link = request.args.get('link')
    return "data:image/jpeg;base64," + makeQR("http://" + link)

    
if __name__ == '__main__':
    table_check()# This code checks whether database table is created or not
    if production: #Check if production variable is set to true use the waitress webserver, else use the buildin flask webserver, with more debug output
        serve(app, host=host, port= 5000, url_scheme=url_scheme) #Start the Webserver for all users on port 5000
    else:
        app.run(host=host, port=5000, debug=True) #Start the Webserver in Debug mode. This means, if the script runs in an error, it will show the error message in Browser.