Skip to content
Snippets Groups Projects
main.py 11.4 KiB
Newer Older
  • Learn to ignore specific revisions
  • Jonas Leder's avatar
    Jonas Leder committed
    #!/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)
    
    Jonas Leder's avatar
    Jonas Leder committed
    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
    
    Jonas Leder's avatar
    Jonas Leder committed
    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
    
    Jonas Leder's avatar
    Jonas Leder committed
    
    app = Flask(__name__)
    
    app.config['GITHUB_CLIENT_ID'] = ''
    app.config['GITHUB_CLIENT_SECRET'] = ''
    github = GitHub(app)
    
    
    Jonas Leder's avatar
    Jonas Leder committed
    domain_to_index = {}
    
    user = ""
    
    Jonas Leder's avatar
    Jonas Leder committed
    
    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()
    
    Jonas Leder's avatar
    Jonas Leder committed
        else:
            builddate = ""
    
    Jonas Leder's avatar
    Jonas Leder committed
    except:
    
    Jonas Leder's avatar
    Jonas Leder committed
        builddate = "" #If the enviorement Variable is not set also skip the builddate
    
    Jonas Leder's avatar
    Jonas Leder committed
    
    
    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()
    
    Jonas Leder's avatar
    Jonas Leder committed
        else:
            version = ""
    
    Jonas Leder's avatar
    Jonas Leder committed
        version = "" #If the enviorement Variable is not set also skip the version
    
    Jonas Leder's avatar
    Jonas Leder committed
    try:
    
    Jonas Leder's avatar
    Jonas Leder committed
        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.
    
    Jonas Leder's avatar
    Jonas Leder committed
        recaptchaPublicKey = os.environ["recaptcha_public"]
    
    Jonas Leder's avatar
    Jonas Leder committed
        if(recaptchaPrivateKey != "") and (recaptchaPublicKey != ""): #If the variables are empty also skip the captcha
            skipCaptcha = False
        else:
            skipCaptcha = True
    
    
    Jonas Leder's avatar
    Jonas Leder committed
    except:
        recaptchaPrivateKey = ""
    
        recaptchaPublicKey = ""
    
    Jonas Leder's avatar
    Jonas Leder committed
        skipCaptcha = True
    
    
    Jonas Leder's avatar
    Jonas Leder committed
    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
    
    
    Jonas Leder's avatar
    Jonas Leder committed
    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.
    
    Jonas Leder's avatar
    Jonas Leder committed
        url_scheme = os.environ["url_scheme"]
    except:
        url_scheme = "http"
    
    
    try:
        host=os.environ["host"]
    except:
        host="127.0.0.1"
    
    
    Jonas Leder's avatar
    Jonas Leder committed
    index = 0
    
    Jonas Leder's avatar
    Jonas Leder committed
    domain_prepared = ""
    
    Jonas Leder's avatar
    Jonas Leder committed
    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:
    
    Jonas Leder's avatar
    Jonas Leder committed
        showDomainSelect=False
    
        domain_prepared = domain[0]
    
    Jonas Leder's avatar
    Jonas Leder committed
    def table_check(): #This function is used on start to make a new Database if not already exists.
    
    Jonas Leder's avatar
    Jonas Leder committed
        create_table = """
            CREATE TABLE WEB_URL(
            LONG_URL TEXT NOT NULL,
            SHORT_URL TEXT NOT NULL
            );
            """
        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
    
    Jonas Leder's avatar
    Jonas Leder committed
        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
    
    Jonas Leder's avatar
    Jonas Leder committed
        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
    
    Jonas Leder's avatar
    Jonas Leder committed
        return response 
    
    
    
    Jonas Leder's avatar
    Jonas Leder committed
    @app.route('/', methods=['GET'])
    def home_get():
    
        return render_template('home.html', builddate=builddate, version=version, domain=domain_prepared, recaptchaPublicKey=recaptchaPublicKey, showDomainSelect=showDomainSelect) #return the default site to create a new shorten link
    
    Jonas Leder's avatar
    Jonas Leder committed
    
    @app.route('/', methods=['POST']) #This function is used to create a new url
    def home_post():
    
    Jonas Leder's avatar
    Jonas Leder committed
        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) #return the user the prefilled form with an error message, because no url to short was provided
    
    Jonas Leder's avatar
    Jonas Leder committed
        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) #return the user the prefilled form with an error message, because no url to short was provided
    
    Jonas Leder's avatar
    Jonas Leder committed
        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) #return the user the prefilled form with an error message, because no short link was provided
    
    Jonas Leder's avatar
    Jonas Leder committed
        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) VALUES (?, ?)',
                    [url, shorturl]
                )
    
                return render_template('home.html', short_url=shorturl, recaptchaPublicKey=recaptchaPublicKey, builddate=builddate, version=version, domain=domain_prepared, qrcode=makeQR("http://" + shorturl)) #return the shorten link to the user
    
    Jonas Leder's avatar
    Jonas Leder committed
            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) #return the user the prefilled form with an error message, because the url was already used
    
    Jonas Leder's avatar
    Jonas Leder committed
        
    @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('/login')
    def login():
        return github.authorize(scope="user")
    
    @app.route('/github-callback')
    @github.authorized_handler
    def authorized(oauth_token):
        if oauth_token is None:
            return "oauth failed, please try again"
        
        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("Hello " + username + "(" + userID + ")")
        resp.set_cookie('userID', userID)
        resp.set_cookie('username', username)
        return resp
        
    
    
    Jonas Leder's avatar
    Jonas Leder committed
    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
    
    Jonas Leder's avatar
    Jonas Leder committed
        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.