freepost-x/freepost/__init__.py

#!./venv/bin/python3

import bleach
import bottle
import configparser
import dateutil.parser
import functools
import hashlib
import importlib
import json
import markdown
import timeago
import urllib3
import yaml
from datetime import datetime, timezone
from bottle import abort, get, jinja2_template as template, post, redirect, request, response, static_file
from urllib.parse import urlparse

# This is used to export the bottle object for a WSGI server
# See passenger_wsgi.py
application = bottle.app ()

# Load user settings for this app
with open ('settings.yaml', encoding='UTF-8') as file:
    settings = yaml.safe_load (file)

# Directories to search for app templates
bottle.TEMPLATE_PATH = [ './freepost/templates' ]

# Custom settings and functions for templates
template = functools.partial (
    template,
    template_settings = {
        'filters': {
            'ago': lambda date: timeago.format(date),
            'datetime': lambda date: date,# date.strftime ('%b %-d, %Y - %H:%M%p%z%Z'),
            # TODO this should be renamed. It's only a way to pretty print dates
            'title': lambda date: dateutil.parser.parse(date).strftime('%b %-d, %Y - %H:%M%z%Z'),
            # Convert markdown to plain text
            'md2txt': lambda text: bleach.clean (markdown.markdown(text),
                                                 tags=[], attributes={}, strip=True),
            # Convert markdown to html
            'md2html': lambda text: bleach.clean (bleach.linkify (markdown.markdown (
                                        text,
                                        # https://python-markdown.github.io/extensions/
                                        extensions=[ 'extra', 'admonition', 'nl2br', 'smarty' ],
                                        output_format='html5'))),
            # Get the domain part of a URL
            'netloc': lambda url: urlparse (url).netloc
        },
        'globals': {
            'new_messages': lambda user_id: database.count_unread_messages (user_id),
            'now': lambda: datetime.now (timezone.utc),
            'request': request,
            # Return a setting from 'settings.yaml'
            'settings': lambda section, key: settings[section][key],
            # Get the current user of the session
            'session_user': lambda: session.user (),
            # Split a string of topics into a list
            'split_topics': lambda topics: [ topic for topic in topics.split (' ') ] if topics else [],
            'url': application.get_url,
        },
        'autoescape': True
    })

# "bleach" library is used to sanitize the HTML output of jinja2's "md2html"
# filter. The library has only a very restrictive list of white-listed
# tags, so we add some more here.
# The list() casting is required because it's of type "frozenlist"
bleach.sanitizer.ALLOWED_TAGS = list(bleach.sanitizer.ALLOWED_TAGS) + [ 'br', 'img', 'p', 'pre', 'h1', 'h2', 'h3', 'hr' ]
bleach.sanitizer.ALLOWED_ATTRIBUTES.update ({
    'img': [ 'src' ]
})

from freepost import database, mail, session

# Decorator.
# Make sure user is logged in
def requires_login (controller):
    def wrapper ():
        session_token = request.get_cookie (
            key    = settings['session']['name'],
            secret = settings['cookies']['secret'])

        if database.is_valid_session (session_token):
            return controller ()
        else:
            redirect (application.get_url ('login'))

    return wrapper

# Decorator.
# Make sure user is logged out
def requires_logout (controller):
    def wrapper ():
        session_token = request.get_cookie (
            key    = settings['session']['name'],
            secret = settings['cookies']['secret'])

        if database.is_valid_session (session_token):
            redirect (application.get_url ('user_settings'))
        else:
            return controller ()

    return wrapper

# Routes

@get ('/', name='homepage')
def homepage ():
    """
    Display homepage with posts sorted by 'hot'.
    """

    # Sort order
    sort = request.query.sort or 'hot'

    # Page number
    page = int (request.query.page or 0)

    if page < 0:
        redirect (application.get_url ('homepage'))

    user = session.user ()

    if sort in [ 'hot', 'new' ]:
        posts = database.get_posts (
                    page, user['id'] if user else None,
                    sort)
    else:
        posts = []

    # Disable browser caching
    # Fix issue: https://notabug.org/zPlus/freepost/issues/80
    response.set_header('cache-control', 'no-cache, no-store, must-revalidate')
    response.set_header('pragma', 'no-cache')
    response.set_header('expires', '0')

    return template ('homepage.html', posts=posts, sort=sort)

# TODO implement this
@get ('/topic/<name>', name='topic')
def topic (name):
    """
    Display posts by topic.
    """

    # Sort order
    sort = request.query.sort or 'hot'

    # Page number
    page = int (request.query.page or 0)

    if page < 0:
        redirect (application.get_url ('homepage'))

    user = session.user ()

    if sort in [ 'hot', 'new' ]:
        posts = database.get_posts (
                    page, user['id'] if user else None,
                    sort, name)
    else:
        posts = []

    return template (
        'homepage.html',
        topic=name,
        posts=posts)

@get ('/about', name='about')
def about ():
    """
    Display "About" page.
    """

    return template ('about.html')

@get ('/login', name='login')
@requires_logout
def login ():
    """
    The login page.
    """

    return template ('login.html')

@post ('/login')
@requires_logout
def login_check ():
    """
    Check login form.
    """

    username = request.forms.getunicode ('username')
    password = request.forms.getunicode ('password')
    remember = 'remember' in request.forms

    if not username or not password:
        return template (
            'login.html',
            flash = 'Bad login!')

    # Check if user exists
    user = database.check_user_credentials (username, password)

    # Username/Password not working
    if not user:
        return template (
            'login.html',
            flash = 'Bad login!')

    # Delete any existing "reset token"
    database.delete_password_reset_token (user['id'])

    # Start new session
    session.start (user['id'], remember)

    # Redirect logged in user to preferred feed
    if user['preferred_feed'] == 'new':
        redirect (application.get_url ('homepage') + '?sort=new')
    else:
        redirect (application.get_url ('homepage'))

@get ('/register', name='register')
@requires_logout
def register ():
    """
    Register new account.
    """

    return template ('register.html')

@post ('/register')
@requires_logout
def register_new_account ():
    """
    Check form for creating new account.
    """

    username = request.forms.getunicode ('username')
    password = request.forms.getunicode ('password')

    # Normalize username
    username = username.strip ()

    # Check if username already exists.
    # Use case-insensitive match to prevent two similar usernames.
    if len (username) == 0 or database.username_exists (username, case_sensitive=False):
        return template (
            'register.html',
            flash='Name taken, please choose another.')

    # Password too short?
    if len (password) < 8:
        return template (
            'register.html',
            flash = 'Password too short')

    # Username OK, Password OK: create new user
    database.new_user (username, password)

    # Retrieve user (check if it was created)
    user = database.check_user_credentials (username, password)

    # Something bad happened...
    if user is None:
        return template (
            'register.html',
            flash = 'An error has occurred, please try again.')

    # Start session...
    session.start (user['id'])

    # ... and go to the homepage of the new user
    redirect (application.get_url ('user_settings'))

@get ('/logout', name='logout')
@requires_login
def logout ():
    """
    Logout user and return to homepage.
    """

    session.close ()

    redirect (application.get_url ('homepage'))

@get ('/password_reset', name='password_reset')
@requires_logout
def password_reset ():
    """
    Display form to reset users password.
    """

    return template ('login_reset.html')

@post ('/password_reset', name='password_reset_send_code')
@requires_logout
def password_reset_send_code ():
    """
    Validate form for resetting password, and if valid send secret
    code via email.
    """

    username = request.forms.getunicode('username')
    email = request.forms.getunicode('email')

    if not username or not email:
        redirect(application.get_url('change_password'))

    user = database.get_user_by_username(username)

    if not user:
        redirect(application.get_url('change_password'))

    # Make sure the given email matches the one that we have in the database
    if user['email'] != email:
        redirect(application.get_url('change_password'))

    # Is there another valid token already (from a previous request)?
    # If yes, do not send another one (to prevent multiple requests or spam)
    if database.is_password_reset_token_valid(user['id']):
        redirect(application.get_url('change_password'))

    # Generate secret token to send via email
    secret_token = random.ascii_string(32)

    # Add token to database
    database.set_password_reset_token(user['id'], secret_token)

    # Send token via email
    client_ip     = request.environ.get('HTTP_X_FORWARDED_FOR') or \
                    request.environ.get('REMOTE_ADDR')
    email_to      = user['email']
    email_subject = 'freepost password reset'
    email_body    = template(
                        'email/password_reset.txt',
                        ip=client_ip,
                        secret_token=secret_token)

    mail.send(email_to, email_subject, email_body)

    redirect(application.get_url('change_password'))

@get ('/change_password', name='change_password')
@requires_logout
def change_password ():
    """
    After the secret code was sent via email, display this form where
    the user can insert the secret code + new password.
    """

    return template ('login_change_password.html')

@post ('/change_password', name='validate_new_password')
@requires_logout
def validate_new_password ():
    """
    Validate the new password, check the secret code, and if everything
    is OK change the user password.
    """

    username     = request.forms.getunicode('username')
    email        = request.forms.getunicode('email')
    password     = request.forms.getunicode('password')
    secret_token = request.forms.getunicode('token')

    # We must have all fields
    if not username or not email or not password or not secret_token:
        redirect(application.get_url('login'))

    # Password too short?
    if len (password) < 8:
        return template (
            'login_change_password.html',
            flash = 'Password must be at least 8 characters long')

    # OK, everything should be fine now. Reset user password.
    database.reset_password(username, email, password, secret_token)

    # Check if the password was successfully reset
    user = database.check_user_credentials (username, password)

    # Username/Password not working
    if not user:
        redirect (application.get_url ('login'))

    # Everything matched!
    # Notify user of password change.
    email_to      = user['email']
    email_subject = 'freepost password changed'
    email_body    = template ('email/password_changed.txt')

    mail.send (email_to, email_subject, email_body)

    # Start new session and redirect user
    session.start (user['id'])
    redirect (application.get_url ('user_settings'))

@get ('/user/settings', name='user_settings')
@requires_login
def user_settings ():
    """
    A user's personal page.
    """

    return template ('user_settings.html')

@post ('/user/settings')
@requires_login
def update_user_settings ():
    """
    Update user info (about, email, ...).
    """

    user = session.user ()

    about = request.forms.getunicode ('about')
    email = request.forms.getunicode ('email')
    preferred_feed = request.forms.getunicode ('preferred_feed')

    if about is None or email is None:
        redirect (application.get_url ('user_settings'))

    if preferred_feed not in [ 'hot', 'new' ]:
        preferred_feed = 'hot'

    # Update user info in the database
    database.update_user (user['id'], about, email, False, preferred_feed)

    redirect (application.get_url ('user_settings'))

@get ('/user_activity/posts')
@requires_login
def user_posts ():
    user = session.user ()
    posts = database.get_user_posts (user['id'])

    return template ('user_posts.html', posts=posts)

@get ('/user_activity/comments')
@requires_login
def user_comments ():
    user = session.user ()
    comments = database.get_user_comments (user['id'])

    return template ('user_comments.html', comments=comments)

@get ('/user_activity/replies', name='user_replies')
@requires_login
def user_replies ():
    user = session.user ()
    replies = database.get_user_replies (user['id'])

    database.set_replies_as_read (user['id'])

    return template ('user_replies.html', replies=replies)

@get ('/user/public/<username>', name='user_public')
def user_public_homepage (username):
    """
    Display a publicly accessible page with public info about the user.
    """

    account = database.get_user_by_username (username)

    if account is None:
        redirect (application.get_url ('user_settings'))

    return template ('user_public_homepage.html', account=account)

@get ('/post/<hash_id>', name='post')
def post_thread (hash_id):
    """
    Display a single post with all its comments.
    """

    user = session.user ()
    post = database.get_post (hash_id, user['id'] if user else None)
    comments = database.get_post_comments (post['id'], user['id'] if user else None)
    topics = database.get_post_topics (post['id'])

    # Group comments by parent
    comments_tree = {}

    for comment in comments:
        if comment['parentId'] is None:
            if 0 not in comments_tree:
                comments_tree[0] = []

            comments_tree[0].append(dict(comment))
        else:
            if comment['parentId'] not in comments_tree:
                comments_tree[comment['parentId']] = []

            comments_tree[comment['parentId']].append(dict(comment))

    # Build ordered list of comments (recourse tree)
    def children (parent_id = 0, depth = 0):
        temp_comments = []

        if parent_id in comments_tree:
            for comment in comments_tree[parent_id]:
                comment['depth'] = depth
                temp_comments.append (comment)

                temp_comments.extend (children (comment['id'], depth + 1))

        return temp_comments

    comments = children ()

    # Show posts/comments Markdown instead of rendered text
    show_source = 'source' in request.query

    # Disable browser caching
    # Fix issue: https://notabug.org/zPlus/freepost/issues/80
    response.set_header('cache-control', 'no-cache, no-store, must-revalidate')
    response.set_header('pragma', 'no-cache')
    response.set_header('expires', '0')

    return template (
        'post.html',
        post=post,
        comments=comments,
        topics=topics,
        show_source=show_source,
        votes = {
            'post': {},
            'comment': {}
        })

@requires_login
@post ('/post/<hash_id>')
def new_comment (hash_id):
    # The comment text
    comment = request.forms.getunicode ('new_comment').strip ()

    # Empty comment?
    if len (comment) == 0:
        redirect (application.get_url ('post', hash_id=hash_id))

    # Retrieve the post
    post = database.get_post (hash_id)

    # Does post exist?
    if not post:
        redirect (application.get_url ('homepage'))

    user = session.user ()

    comment_hash_id = database.new_comment (comment, hash_id, user['id'], post['userId'])

    # Retrieve new comment
    comment = database.get_comment (comment_hash_id)

    # Automatically add an upvote for the original poster
    database.vote_comment (comment['id'], user['id'], +1)

    redirect (application.get_url ('post', hash_id=hash_id))

@requires_login
@get ('/edit/post/<hash_id>', name='edit_post')
def edit_post (hash_id):
    user = session.user ()
    post = database.get_post (hash_id, user['id'])

    # Make sure the session user is the actual poster/commenter
    if post['userId'] != user['id']:
        redirect (application.get_url ('post', hash_id=hash_id))

    return template ('edit_post.html', post=post)

@requires_login
@post ('/edit/post/<hash_id>')
def edit_post_check (hash_id):
    user = session.user ()
    post = database.get_post (hash_id, user['id'])

    # Make sure the session user is the actual poster/commenter
    if post['userId'] != user['id']:
        redirect (application.get_url ('homepage'))

    # MUST have a title. If empty, use original title
    title = request.forms.getunicode ('title').strip ()
    if len (title) == 0: title = post['title']
    link  = request.forms.getunicode ('link').strip () if 'link' in request.forms else ''
    text  = request.forms.getunicode ('text').strip () if 'text' in request.forms else ''

    # If there is a URL, make sure it has a "scheme"
    if len (link) > 0 and urlparse (link).scheme == '':
        link = 'http://' + link

    # Update post
    database.update_post (title, link, text, hash_id, user['id'])

    redirect (application.get_url ('post', hash_id=hash_id))

@requires_login
@get ('/edit/comment/<hash_id>', name='edit_comment')
def edit_comment (hash_id):
    user = session.user ()
    comment = database.get_comment (hash_id, user['id'])

    # Make sure the session user is the actual poster/commenter
    if comment['userId'] != user['id']:
        redirect (application.get_url ('post', hash_id=comment['postHashId']))

    return template ('edit_comment.html', comment=comment)

@requires_login
@post ('/edit/comment/<hash_id>')
def edit_comment_check (hash_id):
    user = session.user ()
    comment = database.get_comment (hash_id, user['id'])

    # Make sure the session user is the actual poster/commenter
    if comment['userId'] != user['id']:
        redirect (application.get_url ('homepage'))

    text = request.forms.getunicode ('text').strip () if 'text' in request.forms else ''

    # Only update if the text is not empty
    if len (text) > 0:
        database.update_comment (text, hash_id, user['id'])

    redirect (application.get_url ('post', hash_id=comment['postHashId']) + '#comment-' + hash_id)

@get ('/submit')
@requires_login
def submit ():
    """
    Submit a new post.
    """

    return template ('submit.html')

@post ('/submit', name='submit')
@requires_login
def submit_check ():
    """
    Check submission of new post.
    """

    # Somebody sent a <form> without a title???
    if not request.forms.getunicode ('title'):
        abort ()

    user = session.user ()

    # Retrieve title
    title = request.forms.getunicode ('title').strip ()

    # Bad title?
    if len (title) == 0:
        return template ('submit.html', flash='Title is missing.')

    # Retrieve link
    link = request.forms.getunicode ('link').strip ()

    if len (link) > 0:
        # If there is a URL, make sure it has a "scheme"
        if urlparse (link).scheme == '':
            link = 'http://' + link

        # Check if this link was already posted in order to avoid duplicate posting.
        previous_posts = database.link_exists(link)
        if previous_posts:
            posts_list = ''.join([
                '<li><a href="{link}">{title}</a></li>'.format(
                    link  = application.get_url ('post', hash_id=post['hashId']),
                    title = post['title'])
                for post in previous_posts ])

            return template ('submit.html',
                             flash='This link was already submitted:<ul>{posts}</ul>'.format(posts=posts_list))

    # Retrieve topics
    topics = request.forms.getunicode ('topics')

    # Retrieve text
    text = request.forms.getunicode ('text')

    # Add the new post
    post_hash_id = database.new_post (title, link, text, user['id'])

    # Retrieve the new post just created
    post = database.get_post (post_hash_id)

    # Add topics for this post
    database.replace_post_topics (post['id'], topics)

    # Automatically add an upvote for the original poster
    database.vote_post (post['id'], user['id'], +1)

    # Posted. Now go the new post's page
    redirect (application.get_url ('post', hash_id=post_hash_id))

@requires_login
@get ('/reply/<hash_id>', name='reply')
def reply (hash_id):
    user = session.user ()

    # The comment to reply to
    comment = database.get_comment (hash_id)

    # Does the comment exist?
    if not comment:
        redirect (application.get_url ('homepage'))

    return template ('reply.html', comment=comment)

@requires_login
@post ('/reply/<hash_id>')
def reply_check (hash_id):
    user = session.user ()

    # The comment to reply to
    comment = database.get_comment (hash_id)

    # The text of the reply
    text = request.forms.getunicode ('text').strip ()

    # Empty comment. Redirect to parent post
    if len (text) == 0:
        redirect (application.get_url ('post', hash_id=comment['postHashId']))

    # We have a text, add the reply and redirect to the new reply
    reply_hash_id = database.new_comment (
                        text, comment['postHashId'], user['id'],
                        comment['userId'], comment['id'])

    # Retrieve new comment
    comment = database.get_comment (reply_hash_id)

    # Automatically add an upvote for the original poster
    database.vote_comment (comment['id'], user['id'], +1)

    redirect (application.get_url ('post', hash_id=comment['postHashId']) + '#comment-' + reply_hash_id)

@requires_login
@post ('/vote', name='vote')
def vote ():
    """
    Handle upvotes and downvotes.
    """

    user = session.user ()

    # Vote a post
    if request.forms.getunicode ('target') == 'post':
        # Retrieve the post
        post = database.get_post (request.forms.getunicode ('post'), user['id'])

        if not post:
            return

        # If user clicked the "upvote" button...
        if request.forms.getunicode ('updown') == 'up':
            # If user has upvoted this post before...
            if post['user_vote'] == 1:
                # Remove +1
                database.vote_post (post['id'], user['id'], -1)
            # If user has downvoted this post before...
            elif post['user_vote'] == -1:
                # Change vote from -1 to +1
                database.vote_post (post['id'], user['id'], +2)
            # If user hasn't voted this post...
            else:
                # Add +1
                database.vote_post (post['id'], user['id'], +1)

        # If user clicked the "downvote" button...
        if request.forms.getunicode ('updown') == 'down':
            # If user has downvoted this post before...
            if post['user_vote'] == -1:
                # Remove -1
                database.vote_post (post['id'], user['id'], +1)
            # If user has upvoted this post before...
            elif post['user_vote'] == 1:
                # Change vote from +1 to -1
                database.vote_post (post['id'], user['id'], -2)
            # If user hasn't voted this post...
            else:
                # Add -1
                database.vote_post (post['id'], user['id'], -1)

    # Vote a comment
    if request.forms.getunicode ('target') == 'comment':
        # Retrieve the comment
        comment = database.get_comment (request.forms.getunicode ('comment'), user['id'])

        if not comment:
            return

        # If user clicked the "upvote" button...
        if request.forms.getunicode ('updown') == 'up':
            # If user has upvoted this comment before...
            if comment['user_vote'] == 1:
                # Remove +1
                database.vote_comment (comment['id'], user['id'], -1)
            # If user has downvoted this comment before...
            elif comment['user_vote'] == -1:
                # Change vote from -1 to +1
                database.vote_comment (comment['id'], user['id'], +2)
            # If user hasn't voted this comment...
            else:
                # Add +1
                database.vote_comment (comment['id'], user['id'], +1)

        # If user clicked the "downvote" button...
        if request.forms.getunicode ('updown') == 'down':
            # If user has downvoted this comment before...
            if comment['user_vote'] == -1:
                # Remove -1
                database.vote_comment (comment['id'], user['id'], +1)
            # If user has upvoted this comment before...
            elif comment['user_vote'] == 1:
                # Change vote from +1 to -1
                database.vote_comment (comment['id'], user['id'], -2)
            # If user hasn't voted this comment...
            else:
                # Add -1
                database.vote_comment (comment['id'], user['id'], -1)

@get ('/search', name='search')
def search ():
    """
    Search content on this instance, and display the results.
    """

    # Get the search query
    query = request.query.getunicode ('q')

    # Get the offset
    page = int (request.query.getunicode ('page') or 0)

    if page < 0:
        return "Page cannot be less than zero."

    # Page increment
    if 'next' in request.query:
        page += 1
    if 'previous' in request.query:
        page -= 1

    # Results order
    sort = request.query.getunicode ('sort')
    if sort not in [ 'newest', 'points' ]:
        sort = 'newest'

    posts = database.search (query, sort=sort, page=page) or []

    return template ('search.html', posts=posts)

@get ('/rss')
def rss_default ():
    """
    Redirect base RSS URL to default "hot" feed.
    """

    return redirect (application.get_url ('rss', sorting='hot'))

@get ('/rss/<sorting>', name='rss')
def rss (sorting):
    """
    Output an RSS feed of posts.
    """

    posts = []

    # Retrieve the hostname from the HTTP request.
    # This is used to build absolute URLs in the RSS feed.
    base_url = request.urlparts.scheme + '://' + request.urlparts.netloc

    if sorting == 'new':
        posts = database.get_posts (sort='new')

    if sorting == 'hot':
        posts = database.get_posts (sort='hot')

    # Set correct Content-Type header for this RSS feed
    response.content_type = 'application/rss+xml; charset=UTF-8'

    return template ('rss.xml', base_url=base_url, posts=posts)

@get ('/rss_comments', name='rss_comments')
def rss_comments ():
    """
    Output an RSS feed of the latest comments.
    """

    # Retrieve the hostname from the HTTP request.
    # This is used to build absolute URLs in the RSS feed.
    base_url = request.urlparts.scheme + '://' + request.urlparts.netloc

    comments = database.get_latest_comments ()

    # Set correct Content-Type header for this RSS feed
    response.content_type = 'application/rss+xml; charset=UTF-8'

    return template ('rss_comments.xml', base_url=base_url, comments=comments)

@get ('/<filename:path>', name='static')
def static (filename):
    """
    Serve static files.
    """

    return static_file (filename, root='freepost/static/')