engineering/remarkbox
1
0
Fork 0
Code Issues 3 Pull requests 1 Projects Releases Packages Wiki Activity Actions

sunset miscutils

Browse source 
modified:   remarkbox/__init__.py
	modified:   remarkbox/lib/mail.py
	modified:   remarkbox/lib/render.py
	new file:   remarkbox/lib/sanitize_html.py
	modified:   remarkbox/models/user.py
	modified:   remarkbox/scripts/json_import.py
	modified:   requirements.py3.txt
	modified:   requirements.txt
This commit is contained in:
Russell Ballestrini 2022-08-01 11:14:10 -04:00
parent d7f8039778
commit 313f9175c5
8 changed files with 401 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

52
remarkbox/__init__.py
View file

@ -31,14 +31,62 @@ from pkg_resources import iter_entry_points
# needed to support expanding ENV vars from ini.
from os.path import expandvars
from miscutils import get_children_settings
import logging
log = logging.getLogger(__name__)
JINJA2_EXTENSION = ".j2"
def get_int_or_bool_or_none_or_str(value):
"""
Given a string value pulled from a configuration file,
this function attempts to return the value with the proper type.
"""
try:
return int(value)
except ValueError:
if value.lower() in {"yes", "y", "true", "y"}:
return True
elif value.lower() in {"no", "n", "false", "f"}:
return False
elif value.lower() == "none":
return None
return str(value)
def get_children_settings(settings, parent_key):
"""
Accept a settings dict and parent key, return dict of children
For example:
auth_tkt.hashalg = md5
Results to:
{'auth_tkt.hashalg': 'md5'}
This function returns the following:
>>> get_children_settings({'auth_tkt.hashalg': 'md5'}, 'auth_tkt')
{'hashalg': 'md5'}
"""
# needed to support expanding ENV vars from ini.
from os.path import expandvars
# the +1 is the . between parent and child settings.
parent_len = len(parent_key) + 1
children = {}
for key, value in settings.items():
if parent_key in key:
# expandvars replaces template with ENV vars.
children[key[parent_len:]] = get_int_or_bool_or_none_or_str(
expandvars(value)
)
return children
def load_entry_points(group_name):
"""Return a dictionary of entry_points related to given group_name"""
entry_points = {}

108
remarkbox/lib/mail.py
View file

@ -1,5 +1,3 @@
from miscutils.mail import send_pyramid_email
# quote email address in OTP so that a plus address
# is not decoded as a space during authentication.
try:
@ -17,6 +15,17 @@ from remarkbox.lib.mail_messages import (
OPERATOR_HTML,
)
import dkim
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
# catch socket errors when postfix isn't running...
from socket import error as socket_error
from jinja2 import Environment, PackageLoader, select_autoescape
jinja2_env = Environment(
@ -25,6 +34,101 @@ jinja2_env = Environment(
)
def send_email(
to_email,
sender_email,
subject,
message_text,
message_html,
relay="localhost",
dkim_private_key_path="",
dkim_selector="",
dkim_signature_algorithm="ed25519-sha256",
):
# the `email` library assumes it is working with string objects.
# the `dkim` library assumes it is working with byte objects.
# this function performs the acrobatics to make them both happy.
if isinstance(message_text, bytes):
# needed for Python 3.
message_text = message_text.decode()
if isinstance(message_html, bytes):
# needed for Python 3.
message_html = message_html.decode()
sender_domain = sender_email.split("@")[-1]
msg = MIMEMultipart("alternative")
msg.attach(MIMEText(message_text, "plain"))
msg.attach(MIMEText(message_html, "html"))
msg["To"] = to_email
msg["From"] = sender_email
msg["Subject"] = subject
try:
# Python 3 libraries expect bytes.
msg_data = msg.as_bytes()
except:
# Python 2 libraries expect strings.
msg_data = msg.as_string()
if dkim_private_key_path and dkim_selector:
# the dkim library uses regex on byte strings so everything
# needs to be encoded from strings to bytes.
with open(dkim_private_key_path) as fh:
dkim_private_key = fh.read()
headers = [b"To", b"From", b"Subject"]
sig = dkim.sign(
message=msg_data,
selector=str(dkim_selector).encode(),
domain=sender_domain.encode(),
privkey=dkim_private_key.encode(),
include_headers=headers,
signature_algorithm=dkim_signature_algorithm.encode(),
)
# add the dkim signature to the email message headers.
# decode the signature back to string_type because later on
# the call to msg.as_string() performs it's own bytes encoding...
msg["DKIM-Signature"] = sig[len("DKIM-Signature: ") :].decode()
try:
# Python 3 libraries expect bytes.
msg_data = msg.as_bytes()
except:
# Python 2 libraries expect strings.
msg_data = msg.as_string()
# TODO: react if connecting to relay (localhost postfix) is a socket error.
s = smtplib.SMTP(relay)
s.sendmail(sender_email, [to_email], msg_data)
s.quit()
return msg
def send_pyramid_email(request, to_email, subject, message_text, message_html):
"""Thin wrapper around `send_email` to customise settings using request object."""
default_sender = "no-reply@{}".format(request.domain)
sender_email = request.app.get("email.sender", default_sender)
subject = "{} | {}".format(subject, request.app.get("email.subject_postfix", request.domain))
relay = request.app.get("email.relay", "localhost")
dkim_private_key_path = request.app.get("email.dkim_private_key_path", "")
dkim_selector = request.app.get("email.dkim_selector", "")
dkim_signature_algorithm = request.app.get("email.dkim_signature_algorithm", "ed25519-sha256")
send_email(
to_email,
sender_email,
subject,
message_text,
message_html,
relay,
dkim_private_key_path,
dkim_selector,
dkim_signature_algorithm,
)
def send_verification_digits_to_email(request, to_email, raw_digits):
"""
Send email with raw_digits a user may pass to verify & authenticate.

2
remarkbox/lib/render.py
View file

@ -1,4 +1,4 @@
from miscutils.sanitize_html import (
from .sanitize_html import (
default_cleaner,
default_tag_acl,
markdown_to_raw_html,

227
remarkbox/lib/sanitize_html.py Normal file
View file

@ -0,0 +1,227 @@
from markdown import markdown
from functools import partial
from collections import defaultdict
from bleach.sanitizer import Cleaner
from bleach.linkifier import LinkifyFilter
from bleach.callbacks import nofollow, target_blank
from bleach_allowlist import markdown_tags, markdown_attrs, all_styles
from bs4 import BeautifulSoup
import miniuri
import logging
log = logging.getLogger(__name__)
def default_tag_acl():
return defaultdict(list)
def default_cleaner(tag_acl=None):
"""
Returns a default Cleaner object.
We use BeautifulSoup to conditionally whitelist or blacklist
tags based on tag attr_name and attr_value pairs.
For example, this will whitelist Mathjax script tags:
tag_acl = {
"script": [
("type", "math/tex; mode=display", "allow"),
],
}
While this example will blacklist Mathjax script tags:
tag_acl = {
"script": [
("type", "math/tex; mode=display", "deny"),
],
}
"""
if tag_acl is None:
tag_acl = {}
maybe_safe_tags = ["pre", "table", "tr", "td"]
tags = maybe_safe_tags + list(tag_acl.keys()) + markdown_tags
attrs = markdown_attrs
attrs["img"].append("width")
attrs["span"] = ["class"]
# Allow both whitelist and blacklist tag_name/attr/attr_value
# to get past bleach.
# We will conditionally filter tags using BeautifulSoup.
for tag_name in tag_acl.keys():
# tag_name, for example "script".
for attr_name, attr_value, allow_or_deny in tag_acl[tag_name]:
if allow_or_deny == "allow":
# attr_name, for example "type".
# attr_value, for example "math/tex; mode=display".
if tag_name not in attrs:
# if tag_name not in attrs, create it as an empty list.
attrs[tag_name] = []
# append the attr_name to the map of approved
# attributes for the given tag_name.
attrs[tag_name].append(attr_name)
cleaner = Cleaner(tags=tags, attributes=attrs)
# doesn't do anything, but i used to be able to pass it via constructor.
# https://github.com/yourcelf/bleach-allowlist/blob/main/bleach_allowlist/bleach_allowlist.py
#cleaner.all_styles=all_styles
# disable link_protection by default.
cleaner.link_protection = False
# an None signifies a relative URI, which should always be whitelisted.
cleaner.whitelist_domains = [None]
# absolute domain, used for turning relative paths into absolute.
cleaner.absolute_domain = ""
# add conditional_whitelist_tags to cleaner object.
cleaner.tag_acl = tag_acl
return cleaner
def markdown_to_raw_html(data, extra_extensions=None):
"""Accepts a markdown string, returns raw unsanitized HTML"""
extensions = [
"markdown.extensions.codehilite",
"markdown.extensions.fenced_code",
]
if extra_extensions is not None:
extensions.extend(extra_extensions)
return markdown(data, extensions=extensions)
def conditional_tag_filter(soup, cleaner):
"""
Use BeautifulSoup `Soup` object to filter out non-whitelisted tags.
"""
# bleach does not have a way to conditionally accept tags whitelists.
# so we built our own way.
log.info(cleaner.tag_acl)
# for each tag_name in the tag_acl ...
for tag_name in cleaner.tag_acl.keys():
# find all the tags in the DOM that match the tag_name, for example "script".
for tag in soup.find_all(tag_name):
# tag is a BeautifulSoup tag object.
# by default, assume we will extract tag (remove it from the DOM).
extract_tag = True
for attr_name, attr_value, allow_or_deny in cleaner.tag_acl[tag_name]:
# tag attr_name:
#
# * "type"
# * "id"
# * "class"
# * ect
#
# tag attr_value:
#
# * "math/tex; mode=display"
# * "<uuid>"
# * ect
#
# allow_or_deny:
#
# * "allow"
# * "deny"
log.info("{} {} {}".format(attr_name, attr_value, allow_or_deny))
if allow_or_deny == "allow":
if tag.attrs.get(attr_name) == attr_value:
# this tag attr_name/attr_value is whitelisted.
# do not extract tag (remove it from the DOM)
extract_tag = False
break
if extract_tag:
log.info(
"Extracting {} tag with attr_name {} and attr_value {}".format(
tag_name, attr_name, attr_value
)
)
# remove this tag.
# this tag attr_name:attr_value is _not_ whitelisted.
tag.extract()
return soup
def protect_links(soup, cleaner):
for a_tag in soup.find_all("a"):
uri = miniuri.Uri(a_tag.attrs.get("href", ""))
if uri.hostname in cleaner.whitelist_domains:
# domain in whitelist or relative URI so remove rel="nofollow".
a_tag.attrs.pop("rel", None)
if uri.hostname is None and cleaner.absolute_domain:
# make relative path absolute!
# TODO: if we hold scheme in Namespace object we can use it
# when building absolute uris. for now assume https.
uri.scheme = "https"
uri.hostname = cleaner.absolute_domain
a_tag.attrs["href"] = str(uri)
elif cleaner.link_protection:
# domain not in whitelist, replace a_tag with "[link removed]".
a_tag.replace_with("[link removed]")
return soup
def clean_raw_html(raw_html, cleaner=None):
"""
Accepts raw HTML and a cleaner object
Returns sanitized HTML as bytes.
"""
if cleaner is None:
cleaner = default_cleaner()
if cleaner.link_protection == False:
cleaner.filters.append(
partial(
LinkifyFilter,
callbacks=[nofollow, target_blank],
skip_tags=["pre", "code"],
)
)
cleaned_html = cleaner.clean(raw_html)
soup = BeautifulSoup(cleaned_html, "html5lib")
# conditionally accept whitelisted tags, filter out the rest.
soup = conditional_tag_filter(soup, cleaner)
# protect links from abuse.
soup = protect_links(soup, cleaner)
return soup.decode(eventual_encoding="utf-8")

10
remarkbox/models/user.py
View file

@ -23,8 +23,6 @@ from .watcher import Watcher
from .notification import NodeEventNotification
from miscutils import generate_password
import logging
log = logging.getLogger(__name__)
@ -35,6 +33,14 @@ except:
from six import u as unicode
def generate_password(size=32):
"""Return a system generated password"""
from random import choice
letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
digits = "0123456789"
pool = letters + digits
return "".join([choice(pool) for i in range(size)])
class UserSurrogate(RBase, Base):
"""

11
remarkbox/scripts/json_import.py
View file

@ -13,8 +13,6 @@ from ..models import (
is_user_name_available,
)
from miscutils import generate_password
from . import base_parser
try:
@ -23,6 +21,15 @@ except:
from six import u as unicode
def generate_password(size=32):
"""Return a system generated password"""
from random import choice
letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
digits = "0123456789"
pool = letters + digits
return "".join([choice(pool) for i in range(size)])
def get_arg_parser():
parser = base_parser("import comments from JSON")
# parser.add_argument('--dry-run', action='store_true', default=False)

3
requirements.py3.txt
View file

@ -43,9 +43,6 @@ ago
# light weight URI attribute parser.
miniuri
# my miscutils checked out via git as an editable package.
git+https://github.com/russellballestrini/miscutils.git#egg=miscutils
# the official Remarkbox "meta" theme.
#git+https://github.com/russellballestrini/remarkbox-theme-meta.git#egg-remarkbox-theme-meta
git+https://git.unturf.com/engineering/remarkbox/remarkbox-theme-meta.git#egg-remarkbox-theme-meta

4
requirements.txt
View file

@ -52,10 +52,6 @@ ago
# light weight URI attribute parser.
miniuri
# my miscutils checked out via git as an editable package.
#-e git+git@github.com:russellballestrini/miscutils.git#egg=miscutils
git+https://github.com/russellballestrini/miscutils.git#egg=miscutils
# Create URI slugs from titles.
python-slugify