Comment system for Hugo
https://labertasche.tuxstash.de/
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
202 lines
7.8 KiB
202 lines
7.8 KiB
3 years ago
|
#!/usr/bin/env python3
|
||
|
# -*- coding: utf-8 -*-
|
||
|
# /**********************************************************************************
|
||
|
# * _author : Domeniko Gentner
|
||
|
# * _mail : code@tuxstash.de
|
||
|
# * _repo : https://git.tuxstash.de/gothseidank/labertasche
|
||
|
# * _license : This project is under MIT License
|
||
|
# *********************************************************************************/
|
||
|
import re
|
||
|
from sys import stderr
|
||
|
from antispam import is_spam as spam, score
|
||
|
from flask import Blueprint, jsonify, request, make_response, redirect
|
||
|
from flask_cors import cross_origin
|
||
|
from sqlalchemy import exc
|
||
|
from labertasche.database import labertasche_db as db
|
||
|
from labertasche.helper import is_valid_json, default_timestamp, check_gravatar, export_location
|
||
|
from labertasche.mail import mail
|
||
|
from labertasche.models import TComments, TLocation, TEmail
|
||
|
from labertasche.settings import Settings
|
||
|
from secrets import compare_digest
|
||
|
|
||
|
|
||
|
# Blueprint
|
||
|
bp_comments = Blueprint("bp_comments", __name__, url_prefix='/comments')
|
||
|
|
||
|
|
||
|
# Route for adding new comments
|
||
|
@bp_comments.route("/new", methods=['POST'])
|
||
|
@cross_origin()
|
||
|
def check_and_insert_new_comment():
|
||
|
if request.method == 'POST':
|
||
|
settings = Settings()
|
||
|
smileys = settings.smileys
|
||
|
addons = settings.addons
|
||
|
sender = mail()
|
||
|
|
||
|
# Check length of content and abort if too long or too short
|
||
|
if request.content_length > 2048:
|
||
|
return make_response(jsonify(status="post-max-length"), 400)
|
||
|
if request.content_length <= 0:
|
||
|
return make_response(jsonify(status="post-min-length"), 400)
|
||
|
|
||
|
# get json from request
|
||
|
new_comment = request.json
|
||
|
|
||
|
# save and sanitize location, nice try, bitch
|
||
|
location = new_comment['location'].strip().replace('.', '')
|
||
|
|
||
|
# Validate json and check length again
|
||
|
if not is_valid_json(new_comment) or \
|
||
3 years ago
|
len(new_comment['content']) < 40 or \
|
||
|
len(new_comment['email']) < 5:
|
||
3 years ago
|
print("too short", file=stderr)
|
||
|
return make_response(jsonify(status='post-invalid-json'), 400)
|
||
|
|
||
|
# Strip any HTML from message body
|
||
|
tags = re.compile('<.*?>')
|
||
|
special = re.compile('[&].*[;]')
|
||
|
content = re.sub(tags, '', new_comment['content']).strip()
|
||
|
content = re.sub(special, '', content).strip()
|
||
|
|
||
|
# Convert smileys
|
||
|
if addons['smileys']:
|
||
|
for key, value in smileys.items():
|
||
|
content = content.replace(key, value)
|
||
|
|
||
3 years ago
|
# Validate replied_to field is integer
|
||
|
replied_to = new_comment['replied_to']
|
||
|
try:
|
||
|
if replied_to:
|
||
|
replied_to = int(replied_to)
|
||
|
|
||
|
# not a valid id at all
|
||
|
except ValueError:
|
||
|
return make_response(jsonify(status="bad-reply"), 400)
|
||
|
|
||
3 years ago
|
# Update values
|
||
|
new_comment.update({"content": content})
|
||
|
new_comment.update({"email": new_comment['email'].strip()})
|
||
|
new_comment.update({"location": location})
|
||
3 years ago
|
new_comment.update({"replied_to": replied_to})
|
||
3 years ago
|
|
||
|
# Check mail
|
||
|
if not sender.validate(new_comment['email']):
|
||
|
return make_response(jsonify(status='post-invalid-email'), 400)
|
||
|
|
||
|
# check for spam
|
||
|
is_spam = spam(new_comment['content'])
|
||
|
has_score = score(new_comment['content'])
|
||
|
|
||
|
# Insert mail into spam if detected, allow if listed as such
|
||
|
email = db.session.query(TEmail).filter(TEmail.email == new_comment['email']).first()
|
||
|
if not email:
|
||
|
if is_spam:
|
||
|
entry = {
|
||
|
"email": new_comment['email'],
|
||
|
"is_blocked": True,
|
||
|
"is_allowed": False
|
||
|
}
|
||
|
db.session.add(TEmail(**entry))
|
||
|
if email:
|
||
|
if not email.is_allowed:
|
||
|
is_spam = True
|
||
|
if email.is_allowed:
|
||
3 years ago
|
# This forces the comment to be not spam if the address is in the allowed list,
|
||
|
# but the commenter will still need to confirm it to avoid brute
|
||
|
# force attacks against this feature
|
||
3 years ago
|
is_spam = False
|
||
|
|
||
|
# Look for location
|
||
3 years ago
|
loc_query = db.session.query(TLocation) \
|
||
|
.filter(TLocation.location == new_comment['location'])
|
||
3 years ago
|
|
||
|
if loc_query.first():
|
||
3 years ago
|
# Location exists, set existing location id
|
||
3 years ago
|
new_comment.update({'location_id': loc_query.first().id_location})
|
||
|
# TComments does not have this field
|
||
|
new_comment.pop("location")
|
||
|
else:
|
||
|
# Insert new location
|
||
|
loc_table = {
|
||
|
'location': new_comment['location']
|
||
|
}
|
||
|
new_loc = TLocation(**loc_table)
|
||
|
db.session.add(new_loc)
|
||
|
db.session.flush()
|
||
|
db.session.refresh(new_loc)
|
||
|
new_comment.update({'location_id': new_loc.id_location})
|
||
|
|
||
|
# TComments does not have this field
|
||
|
new_comment.pop("location")
|
||
|
|
||
|
# insert comment
|
||
3 years ago
|
# noinspection PyBroadException
|
||
3 years ago
|
try:
|
||
|
new_comment.update({"is_published": False})
|
||
|
new_comment.update({"created_on": default_timestamp()})
|
||
|
new_comment.update({"is_spam": is_spam})
|
||
|
new_comment.update({"spam_score": has_score})
|
||
|
new_comment.update({"gravatar": check_gravatar(new_comment['email'])})
|
||
|
t_comment = TComments(**new_comment)
|
||
|
db.session.add(t_comment)
|
||
|
db.session.commit()
|
||
|
db.session.flush()
|
||
|
db.session.refresh(t_comment)
|
||
|
|
||
|
# Send confirmation link and store returned value
|
||
|
hashes = sender.send_confirmation_link(new_comment['email'])
|
||
|
setattr(t_comment, "confirmation", hashes[0])
|
||
|
setattr(t_comment, "deletion", hashes[1])
|
||
|
db.session.commit()
|
||
|
|
||
|
except exc.IntegrityError as e:
|
||
|
# Comment body exists, because content is unique
|
||
|
print(f"Duplicate from {request.environ['REMOTE_ADDR']}, error is:\n{e}", file=stderr)
|
||
|
return make_response(jsonify(status="post-duplicate"), 400)
|
||
|
|
||
3 years ago
|
except Exception: # must be at bottom
|
||
3 years ago
|
return make_response(jsonify(status="post-internal-server-error"), 400)
|
||
|
|
||
3 years ago
|
export_location(t_comment.location_id)
|
||
3 years ago
|
return make_response(jsonify(status="post-success", comment_id=t_comment.comments_id), 200)
|
||
|
|
||
|
|
||
|
# Route for confirming comments
|
||
|
@bp_comments.route("/confirm/<email_hash>", methods=['GET'])
|
||
|
@cross_origin()
|
||
|
def check_confirmation_link(email_hash):
|
||
|
settings = Settings()
|
||
|
comment = db.session.query(TComments).filter(TComments.confirmation == email_hash).first()
|
||
|
if comment:
|
||
|
location = db.session.query(TLocation).filter(TLocation.id_location == comment.location_id).first()
|
||
|
if compare_digest(comment.confirmation, email_hash):
|
||
|
comment.confirmation = None
|
||
|
if not comment.is_spam:
|
||
|
setattr(comment, "is_published", True)
|
||
|
db.session.commit()
|
||
|
url = f"{settings.system['blog_url']}{location.location}#comment_{comment.comments_id}"
|
||
3 years ago
|
export_location(location.id_location)
|
||
3 years ago
|
return redirect(url)
|
||
|
|
||
|
return redirect(f"{settings.system['blog_url']}?cnf=true")
|
||
|
|
||
|
|
||
|
# Route for deleting comments
|
||
|
@bp_comments.route("/delete/<email_hash>", methods=['GET'])
|
||
|
@cross_origin()
|
||
|
def check_deletion_link(email_hash):
|
||
|
settings = Settings()
|
||
|
query = db.session.query(TComments).filter(TComments.deletion == email_hash)
|
||
|
comment = query.first()
|
||
|
if comment:
|
||
|
location = db.session.query(TLocation).filter(TLocation.id_location == comment.location_id).first()
|
||
|
if compare_digest(comment.deletion, email_hash):
|
||
|
query.delete()
|
||
|
db.session.commit()
|
||
|
url = f"{settings.system['blog_url']}?deleted=true"
|
||
3 years ago
|
export_location(location.id_location)
|
||
3 years ago
|
return redirect(url)
|
||
|
|
||
|
return redirect(f"{settings.system['blog_url']}?cnf=true")
|