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, jsonify #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
import json #used for github oauth
from html import escape #This is used to escape characters, if they are send in the url
from itsdangerous import URLSafeSerializer #used for signing the cookies
import random #used for signing the cookies
import string #used for signing the cookies

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 = ["localhost:5000", "127.0.0.1:500"]

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:
    GITHUB_CLIENT_ID = os.environ['GITHUB_CLIENT_ID']
    GITHUB_CLIENT_SECRET = os.environ['GITHUB_CLIENT_SECRET']
    GOOGLE_CLIENT_ID = os.environ['GOOGLE_CLIENT_ID']
    GOOGLE_CLIENT_SECRET = os.environ['GOOGLE_CLIENT_SECRET']
except:
    print("please set the oauth keys and run again.")
    exit()

try:
    if(os.environ["cookieNotice"] == 1):
        cookieNotice = True
    else:
        cookieNotice = False
except:
    cookieNotice = True

try:
    secretKey = open("db/secretKey.txt", "r").read()
except:
    secretKey = ''.join(random.choice(string.ascii_lowercase) for i in range(100)) #If we can't find the secret key(first run) we generate it in this step and write it to a file
    print("generated secret Key. Key is: " + secretKey)
    f = open("db/secretKey.txt", "w")
    f.write(secretKey)
    f.close()
    secretKey = open("db/secretKey.txt", "r").read()
s = URLSafeSerializer(secretKey)
sAPI = URLSafeSerializer("api_key_" + secretKey)

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
    if(skipCaptcha): return True #If recaptcha is disabled alwas return at this point true, which means response is verified
    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 " + s.loads(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="#" onClick="showLogin()" style="color:white">login</a>'

    return render_template('home.html', builddate=builddate, version=version, domain=domain_prepared, recaptchaPublicKey=recaptchaPublicKey, showDomainSelect=showDomainSelect, loginbar=loginbar, cookieNotice=cookieNotice) #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 = s.loads(request.cookies.get('userID'))
        loginbar = "Hello " + s.loads(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):
        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, cookieNotice=cookieNotice) #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, cookieNotice=cookieNotice) #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, cookieNotice=cookieNotice) #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(url_scheme + "://" + shorturl), loginbar=loginbar, cookieNotice=cookieNotice) #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, cookieNotice=cookieNotice) #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():
    service = request.args.get("service")
    if(service == "github"): return redirect("https://github.com/login/oauth/authorize/?client_id=" + GITHUB_CLIENT_ID + "&scope=user") #redirect the user to the github login page and ask for access to user data (name, email, ...)
    if(service == "google"): return redirect("https://accounts.google.com/o/oauth2/v2/auth?client_id=" + GOOGLE_CLIENT_ID + "&scope=profile%20email%20openid&response_type=code&access_type=offline&include_granted_scopes=true&redirect_uri=" + url_scheme + "://" + domain[0] + "/user/google-callback")
    return render_template("login.html", cookieNotice=cookieNotice)

@app.route("/user/google-callback")
def authorizeGoogle():
    try:
        code = request.args.get("code")
        url = "https://www.googleapis.com/oauth2/v4/token" #The baseurl
        headers = {'Content-Type': 'application/x-www-form-urlencoded',}
        params = {'client_id': GOOGLE_CLIENT_ID, 'client_secret': GOOGLE_CLIENT_SECRET, 'code': code, "grant_type": "authorization_code", "redirect_uri": url_scheme + "://" + domain[0] + "/user/google-callback"} #As paramtere we send the client id and the client secret which we get from github when registering an application and the user code from before
        access_token = post(url,params, headers=headers).text.split('access_token": "')[1].split('"')[0]
        authorization_header = {"Authorization": "OAuth %s" % access_token}
        r = get("https://www.googleapis.com/oauth2/v2/userinfo", headers=authorization_header)
        userID = r.text.split('"id": "')[1].split('"')[0]
        name = r.text.split('"name": "')[1].split('"')[0]
        resp = make_response(redirect('/')) #redirect the user at the end back to the main page
        resp.set_cookie('userID', s.dumps("google_" + userID)) #set the cookies with username and userid
        resp.set_cookie('username', s.dumps(name))
        return resp
    except:
        return "Authentication failed"


@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
def authorizeGithub():
    try:
        code = request.args.get("code")
        url = "https://github.com/login/oauth/access_token" #The baseurl
        params = {'client_id': GITHUB_CLIENT_ID, 'client_secret': GITHUB_CLIENT_SECRET, 'code': code} #As paramtere we send the client id and the client secret which we get from github when registering an application and the user code from before
        oauth_token = post(url,params).text.split("access_token=")[1].split("&")[0] #Send a post request with the parameters from

        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', s.dumps("github_" + userID)) #set the cookies with username and userid
        resp.set_cookie('username', s.dumps(username))
        return resp
    except:
        return "Authentication failed"
    
@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 = s.loads(request.cookies.get('userID')) #Get the userid from the cookie
        loginbar = "Hello " + s.loads(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.' #If user has no shorten links make this message
        except:
            abort(500) #Shouldn't happen, 500 means internal server error
        return render_template('editEntries.html', content=response, loginbar=loginbar, cookieNotice=cookieNotice) #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 = s.loads(request.cookies.get('userID')) #get the userid from the cookie
        loginbar = "Hello " + s.loads(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(url_scheme + "://" + link)

@app.route('/user/api', methods=['POST'])
def api():
    username = "error"
    try:
        username = sAPI.loads(request.form['apikey'])
    except:
        pass

    try:
        short = request.form['short']
    except:
        return jsonify(
            status="1",
            message="short link missing"
        )
    
    try:
        longURL = request.form['long']
    except:
        return jsonify(
            status="2",
            message="link to short is missing"
        )

    domain_ = short.split("/")[0]
    if not domain_ in domain:
        return jsonify(
            status="3",
            message="domain for short link is not in allowed domain list"
        )
    
    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=?', [short])
        try:
            short2 = res.fetchone()
            already_used = False
            if short2 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 (?, ?, ?)',
                [longURL, short, username]
            )

            try:
                request.form['qr']
                qr64 = "data:image/jpeg;base64," + makeQR(url_scheme + "://" + short)
                
                return jsonify(
                    status="0",
                    message="ok",
                    qr=qr64
                )
            except:
                return jsonify(
                    status="0",
                    message="ok"
                )
        else:
            return jsonify(
                status="4",
                message="short url already in use"
            )

@app.route('/user/api', methods=['GET'])
def apiDocs():
    try:
        userID = sAPI.dumps(s.loads(request.cookies.get('userID')))
    except:
        userID = ""

    return render_template("apiDocs.html", apikey=userID, domain=request.headers['Host'], url_scheme=url_scheme)

    
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.