Merge branch 'main' of gitlab.com:bashrc2/epicyon

announce.py

@@ -238,7 +238,7 @@ def announce_public(session, base_dir: str, federation_list: [],
 
 
 def send_announce_via_server(base_dir: str, session,
-                             fromNickname: str, password: str,
+                             from_nickname: str, password: str,
                              from_domain: str, fromPort: int,
                              http_prefix: str, repeat_object_url: str,
                              cached_webfingers: {}, person_cache: {},
@@ -253,7 +253,7 @@ def send_announce_via_server(base_dir: str, session,
     from_domain_full = get_full_domain(from_domain, fromPort)
 
     to_url = 'https://www.w3.org/ns/activitystreams#Public'
-    actor_str = local_actor_url(http_prefix, fromNickname, from_domain_full)
+    actor_str = local_actor_url(http_prefix, from_nickname, from_domain_full)
     cc_url = actor_str + '/followers'
 
     status_number, published = get_status_number()
@@ -270,7 +270,7 @@ def send_announce_via_server(base_dir: str, session,
         'type': 'Announce'
     }
 
-    handle = http_prefix + '://' + from_domain_full + '/@' + fromNickname
+    handle = http_prefix + '://' + from_domain_full + '/@' + from_nickname
 
     # lookup the inbox for the To handle
     wf_request = webfinger_handle(session, handle, http_prefix,
@@ -296,7 +296,7 @@ def send_announce_via_server(base_dir: str, session,
                                   base_dir, session, wf_request,
                                   person_cache,
                                   project_version, http_prefix,
-                                  fromNickname, from_domain,
+                                  from_nickname, from_domain,
                                   post_to_box, 73528)
 
     if not inbox_url:
@@ -309,7 +309,7 @@ def send_announce_via_server(base_dir: str, session,
             print('DEBUG: announce no actor was found for ' + handle)
         return 4
 
-    auth_header = create_basic_auth_header(fromNickname, password)
+    auth_header = create_basic_auth_header(from_nickname, password)
 
     headers = {
         'host': from_domain,

blog.py

@@ -259,7 +259,8 @@ def _html_blog_post_content(debug: bool, session, authorized: bool,
     mute_str = ''
     is_muted = False
     attachment_str, _ = \
-        get_post_attachments_as_html(post_json_object,
+        get_post_attachments_as_html(base_dir, domain_full,
+                                     post_json_object,
                                      'tlblogs', translate,
                                      is_muted, avatar_link,
                                      reply_str, announce_str,

cache.py

@@ -143,6 +143,8 @@ def get_person_pub_key(base_dir: str, session, person_url: str,
                        domain: str, onion_domain: str,
                        i2p_domain: str,
                        signing_priv_key_pem: str) -> str:
+    """Get the public key for an actor
+    """
     if not person_url:
         return None
     person_url = person_url.replace('#main-key', '')

content.py

@@ -1681,3 +1681,40 @@ def create_edits_html(edits_json: {}, post_json_object: {},
     return '<details><summary class="cw">' + \
         translate['SHOW EDITS'] + '</summary>' + \
         edits_str + '</details>'
+
+
+def remove_script(content: str, log_filename: str,
+                  actor: str, url: str) -> str:
+    """Removes <script> from some content
+    """
+    separators = [['<', '>'], ['&lt;', '&gt;']]
+    for sep in separators:
+        prefix = sep[0] + 'script'
+        ending = '/script' + sep[1]
+        if prefix in content:
+            sections = content.split(prefix)
+            ctr = 0
+            for text in sections:
+                if ctr == 0:
+                    ctr += 1
+                    continue
+                if ending not in text:
+                    if '/' + sep[1] not in text:
+                        continue
+                if ending in text:
+                    text = prefix + text.split(ending)[0] + ending
+                else:
+                    text = prefix + text.split('/' + sep[1])[0] + '/' + sep[1]
+                    if log_filename and actor:
+                        # write the detected script to a log file
+                        log_str = actor + ' ' + url + ' ' + text + '\n'
+                        writeType = 'a+'
+                        if os.path.isfile(log_filename):
+                            writeType = 'w+'
+                        try:
+                            with open(log_filename, writeType) as fp_log:
+                                fp_log.write(log_str)
+                        except OSError:
+                            print('EX: cannot append to svg script log')
+                content = content.replace(text, '')
+    return content

happening.py

@@ -1319,7 +1319,8 @@ def dav_month_via_server(session, http_prefix: str,
         '  </c:filter>\n' + \
         '</c:calendar-query>'
     result = \
-        get_method("REPORT", xml_str, session, url, params, headers, debug)
+        get_method("REPORT", xml_str, session, url, params, headers, debug,
+                   __version__, http_prefix, domain)
     return result
 
 
@@ -1365,5 +1366,6 @@ def dav_day_via_server(session, http_prefix: str,
         '  </c:filter>\n' + \
         '</c:calendar-query>'
     result = \
-        get_method("REPORT", xml_str, session, url, params, headers, debug)
+        get_method("REPORT", xml_str, session, url, params, headers, debug,
+                   __version__, http_prefix, domain)
     return result

inbox.py

@@ -70,6 +70,7 @@ from categories import set_hashtag_category
 from httpsig import get_digest_algorithm_from_headers
 from httpsig import verify_post_headers
 from session import create_session
+from session import download_image
 from follow import follower_approval_active
 from follow import is_following_actor
 from follow import get_followers_of_actor
@@ -127,6 +128,95 @@ from webapp_hashtagswarm import html_hash_tag_swarm
 from person import valid_sending_actor
 from fitnessFunctions import fitness_performance
 from content import valid_url_lengths
+from content import remove_script
+
+
+def cache_svg_images(session, base_dir: str, http_prefix: str,
+                     nickname: str, domain: str, domain_full: str,
+                     onion_domain: str, i2p_domain: str,
+                     post_json_object: {},
+                     federation_list: [], debug: bool,
+                     test_image_filename: str) -> bool:
+    """Creates a local copy of a remote svg file
+    """
+    if has_object_dict(post_json_object):
+        obj = post_json_object['object']
+    else:
+        obj = post_json_object
+    if not obj.get('id'):
+        return False
+    if not obj.get('attachment'):
+        return False
+    if not isinstance(obj['attachment'], list):
+        return False
+    cached = False
+    post_id = remove_id_ending(obj['id']).replace('/', '--')
+    actor = 'unknown'
+    if obj.get('attributedTo'):
+        actor = obj['attributedTo']
+    log_filename = base_dir + '/accounts/svg_scripts_log.txt'
+    for index in range(len(obj['attachment'])):
+        attach = obj['attachment'][index]
+        if not attach.get('mediaType'):
+            continue
+        if not attach.get('url'):
+            continue
+        if attach['url'].endswith('.svg') or \
+           'svg' in attach['mediaType']:
+            url = attach['url']
+            if not url_permitted(url, federation_list):
+                continue
+            # if this is a local image then it has already been
+            # validated on upload
+            if '://' + domain in url:
+                continue
+            if onion_domain:
+                if '://' + onion_domain in url:
+                    continue
+            if i2p_domain:
+                if '://' + i2p_domain in url:
+                    continue
+            if '/' in url:
+                filename = url.split('/')[-1]
+            else:
+                filename = url
+            if not test_image_filename:
+                image_filename = \
+                    base_dir + '/media/' + post_id + '_' + filename
+                if not download_image(session, base_dir, url,
+                                      image_filename, debug):
+                    continue
+            else:
+                image_filename = test_image_filename
+            image_data = None
+            try:
+                with open(image_filename, 'rb') as fp_svg:
+                    image_data = fp_svg.read()
+            except OSError:
+                print('EX: unable to read svg file data')
+            if image_data:
+                image_data = image_data.decode()
+                cleaned_up = \
+                    remove_script(image_data, log_filename, actor, url)
+                if cleaned_up != image_data:
+                    # write the cleaned up svg image
+                    svg_written = False
+                    cleaned_up = cleaned_up.encode('utf-8')
+                    try:
+                        with open(image_filename, 'wb') as im_file:
+                            im_file.write(cleaned_up)
+                            svg_written = True
+                    except OSError:
+                        print('EX: unable to write cleaned up svg ' + url)
+                    if svg_written:
+                        # change the url to be the local version
+                        obj['attachment'][index]['url'] = \
+                            http_prefix + '://' + domain_full + '/media/' + \
+                            post_id + '_' + filename
+                        cached = True
+                else:
+                    cached = True
+    return cached
 
 
 def _store_last_post_id(base_dir: str, nickname: str, domain: str,
@@ -4020,6 +4110,15 @@ def _inbox_after_initial(server, inbox_start_time,
         fitness_performance(inbox_start_time, server.fitness,
                             'INBOX', '_obtain_avatar_for_reply_post',
                             debug)
+
+        # cache any svg image attachments locally
+        # This is so that any scripts can be removed
+        cache_svg_images(session, base_dir, http_prefix,
+                         nickname, domain, domain_full,
+                         onion_domain, i2p_domain,
+                         post_json_object,
+                         federation_list, debug, None)
+
         inbox_start_time = time.time()
 
         # save the post to file
@@ -4649,7 +4748,7 @@ def _receive_follow_request(session, session_onion, session_i2p,
                                   message_json['actor'],
                                   person_cache, debug, project_version,
                                   curr_http_prefix,
-                                  domain_to_follow, onion_domain,
+                                  this_domain, onion_domain,
                                   i2p_domain, signing_priv_key_pem):
             if debug:
                 print('Unable to obtain following actor: ' +
@@ -4693,7 +4792,7 @@ def _receive_follow_request(session, session_onion, session_i2p,
             if not get_person_pub_key(base_dir, curr_session,
                                       message_json['actor'],
                                       person_cache, debug, project_version,
-                                      curr_http_prefix, domain_to_follow,
+                                      curr_http_prefix, this_domain,
                                       onion_domain, i2p_domain,
                                       signing_priv_key_pem):
                 if debug:

posts.py

@@ -2802,10 +2802,17 @@ def send_signed_json(post_json_object: {}, session, base_dir: str,
     if debug:
         print('DEBUG: handle - ' + handle + ' to_port ' + str(to_port))
 
+    # domain shown in the user agent
+    ua_domain = curr_domain
+    if to_domain.endswith('.onion'):
+        ua_domain = onion_domain
+    elif to_domain.endswith('.i2p'):
+        ua_domain = i2p_domain
+
     # lookup the inbox for the To handle
     wf_request = webfinger_handle(session, handle, http_prefix,
                                   cached_webfingers,
-                                  domain, project_version, debug,
+                                  ua_domain, project_version, debug,
                                   group_account, signing_priv_key_pem)
     if not wf_request:
         if debug:
@@ -3280,7 +3287,8 @@ def send_to_named_addresses_thread(server, session, session_onion, session_i2p,
 
 
 def _has_shared_inbox(session, http_prefix: str, domain: str,
-                      debug: bool, signing_priv_key_pem: str) -> bool:
+                      debug: bool, signing_priv_key_pem: str,
+                      ua_domain: str) -> bool:
     """Returns true if the given domain has a shared inbox
     This tries the new and the old way of webfingering the shared inbox
     """
@@ -3290,7 +3298,7 @@ def _has_shared_inbox(session, http_prefix: str, domain: str,
     try_handles.append('inbox@' + domain)
     for handle in try_handles:
         wf_request = webfinger_handle(session, handle, http_prefix, {},
-                                      domain, __version__, debug, False,
+                                      ua_domain, __version__, debug, False,
                                       signing_priv_key_pem)
         if wf_request:
             if isinstance(wf_request, dict):
@@ -3404,9 +3412,16 @@ def send_to_followers(server, session, session_onion, session_i2p,
                 curr_session = session_i2p
                 curr_http_prefix = 'http'
 
+        # get the domain showin by the user agent
+        ua_domain = domain
+        if follower_domain.endswith('.onion'):
+            ua_domain = onion_domain
+        elif follower_domain.endswith('.i2p'):
+            ua_domain = i2p_domain
+
         with_shared_inbox = \
             _has_shared_inbox(curr_session, curr_http_prefix, follower_domain,
-                              debug, signing_priv_key_pem)
+                              debug, signing_priv_key_pem, ua_domain)
         if debug:
             if with_shared_inbox:
                 print(follower_domain + ' has shared inbox')

session.py

@@ -58,6 +58,8 @@ def create_session(proxy_type: str):
 
 def url_exists(session, url: str, timeout_sec: int = 3,
                http_prefix: str = 'https', domain: str = 'testdomain') -> bool:
+    """Is the given url resolvable?
+    """
     if not isinstance(url, str):
         print('url: ' + str(url))
         print('ERROR: url_exists failed, url should be a string')
@@ -256,8 +258,7 @@ def get_json(signing_priv_key_pem: str,
 
 def get_vcard(xml_format: bool,
               session, url: str, params: {}, debug: bool,
-              version: str = __version__, http_prefix: str = 'https',
-              domain: str = 'testdomain',
+              version: str, http_prefix: str, domain: str,
               timeout_sec: int = 20, quiet: bool = False) -> {}:
     if not isinstance(url, str):
         if debug and not quiet:
@@ -335,8 +336,7 @@ def get_vcard(xml_format: bool,
 
 def download_html(signing_priv_key_pem: str,
                   session, url: str, headers: {}, params: {}, debug: bool,
-                  version: str = __version__, http_prefix: str = 'https',
-                  domain: str = 'testdomain',
+                  version: str, http_prefix: str, domain: str,
                   timeout_sec: int = 20, quiet: bool = False) -> {}:
     if not isinstance(url, str):
         if debug and not quiet:
@@ -375,8 +375,7 @@ def download_html(signing_priv_key_pem: str,
 
 def download_ssml(signing_priv_key_pem: str,
                   session, url: str, headers: {}, params: {}, debug: bool,
-                  version: str = __version__, http_prefix: str = 'https',
-                  domain: str = 'testdomain',
+                  version: str, http_prefix: str, domain: str,
                   timeout_sec: int = 20, quiet: bool = False) -> {}:
     if not isinstance(url, str):
         if debug and not quiet:
@@ -735,9 +734,10 @@ def download_image_any_mime_type(session, url: str,
 
 def get_method(method_name: str, xml_str: str,
                session, url: str, params: {}, headers: {}, debug: bool,
-               version: str = __version__, http_prefix: str = 'https',
-               domain: str = 'testdomain',
+               version: str, http_prefix: str, domain: str,
                timeout_sec: int = 20, quiet: bool = False) -> {}:
+    """Part of the vcard interface
+    """
     if method_name not in ("REPORT", "PUT", "PROPFIND"):
         print("Unrecognized method: " + method_name)
         return None

tests.py

@@ -128,7 +128,9 @@ from delete import send_delete_via_server
 from inbox import json_post_allows_comments
 from inbox import valid_inbox
 from inbox import valid_inbox_filenames
+from inbox import cache_svg_images
 from categories import guess_hashtag_category
+from content import remove_script
 from content import create_edits_html
 from content import content_diff
 from content import bold_reading_string
@@ -3978,6 +3980,8 @@ def _test_danger_svg(base_dir: str) -> None:
         '  <circle cx="5" cy="5" r="4" />' + \
         '</svg>'
     assert not dangerous_svg(svg_content, False)
+    cleaned_up = remove_script(svg_content, None, None, None)
+    assert cleaned_up == svg_content
     svg_content = \
         '  <svg viewBox="0 0 10 10" xmlns="http://www.w3.org/2000/svg">' + \
         '  <script>' + \
@@ -3999,6 +4003,61 @@ def _test_danger_svg(base_dir: str) -> None:
         '</svg>'
     assert dangerous_svg(svg_content, False)
 
+    svg_clean = \
+        '  <svg viewBox="0 0 10 10" xmlns="http://www.w3.org/2000/svg">' + \
+        '    <circle cx="5" cy="5" r="4" />' + \
+        '</svg>'
+
+    cleaned_up = remove_script(svg_content, None, None, None)
+    assert '<script' not in cleaned_up
+    assert '/script>' not in cleaned_up
+    if cleaned_up != svg_clean:
+        print(cleaned_up)
+    assert cleaned_up == svg_clean
+
+    session = None
+    http_prefix = 'https'
+    nickname = 'amplifier'
+    domain = 'ratsratsrats.live'
+    domain_full = domain
+    onion_domain = None
+    i2p_domain = None
+    federation_list = []
+    debug = True
+    svg_image_filename = base_dir + '/.unit_test_safe.svg'
+    post_json_object = {
+        "object": {
+            "id": "1234",
+            "attributedTo": "someactor",
+            "attachment": [
+                {
+                    "mediaType": "svg",
+                    "url": "https://somesiteorother.net/media/wibble.svg"
+                }
+            ]
+        }
+    }
+
+    with open(svg_image_filename, 'wb+') as fp_svg:
+        fp_svg.write(svg_content.encode('utf-8'))
+    assert os.path.isfile(svg_image_filename)
+    assert svg_content != svg_clean
+
+    assert cache_svg_images(session, base_dir, http_prefix,
+                            nickname, domain, domain_full,
+                            onion_domain, i2p_domain,
+                            post_json_object,
+                            federation_list, debug,
+                            svg_image_filename)
+
+    url = post_json_object['object']['attachment'][0]['url']
+    assert url == 'https://ratsratsrats.live/media/1234_wibble.svg'
+
+    with open(svg_image_filename, 'rb') as fp_svg:
+        cached_content = fp_svg.read().decode()
+    os.remove(svg_image_filename)
+    assert cached_content == svg_clean
+
     assert not scan_themes_for_scripts(base_dir)
 
 

webapp_create_post.py

@@ -379,8 +379,6 @@ def html_new_post(css_cache: {}, media_instance: bool, translate: {},
     new_post_image_section += \
         '      <input type="file" id="attachpic" name="attachpic"'
     formats_string = get_media_formats()
-    # remove svg as a permitted format
-    formats_string = formats_string.replace(', .svg', '').replace('.svg, ', '')
     new_post_image_section += \
         '            accept="' + formats_string + '">\n'
     new_post_image_section += \

webapp_post.py

@@ -381,6 +381,10 @@ def _get_avatar_image_html(showAvatarOptions: bool,
                            page_number: int, message_id_str: str) -> str:
     """Get html for the avatar image
     """
+    # don't use svg images
+    if avatar_url.endswith('.svg'):
+        avatar_url = '/icons/avatar_default.png'
+
     avatar_link = ''
     if '/users/news/' not in avatar_url:
         avatar_link = \
@@ -1962,7 +1966,9 @@ def individual_post_as_html(signing_priv_key_pem: str,
     _log_post_timing(enable_timing_log, post_start_time, '14')
 
     attachment_str, gallery_str = \
-        get_post_attachments_as_html(post_json_object, box_name, translate,
+        get_post_attachments_as_html(base_dir, domain_full,
+                                     post_json_object,
+                                     box_name, translate,
                                      is_muted, avatar_link,
                                      reply_str, announce_str, like_str,
                                      bookmark_str, delete_str, mute_str)

webapp_utils.py

@@ -11,6 +11,7 @@ import os
 from shutil import copyfile
 from collections import OrderedDict
 from session import get_json
+from utils import remove_id_ending
 from utils import get_attachment_property_value
 from utils import is_account_dir
 from utils import remove_html
@@ -1096,7 +1097,8 @@ def _is_attached_video(attachment_filename: str) -> bool:
     return False
 
 
-def get_post_attachments_as_html(post_json_object: {}, box_name: str,
+def get_post_attachments_as_html(base_dir: str, domain_full: str,
+                                 post_json_object: {}, box_name: str,
                                  translate: {},
                                  is_muted: bool, avatar_link: str,
                                  reply_str: str, announce_str: str,
@@ -1116,6 +1118,10 @@ def get_post_attachments_as_html(post_json_object: {}, box_name: str,
     attachment_ctr = 0
     attachment_str = ''
     media_style_added = False
+    post_id = None
+    if post_json_object['object'].get('id'):
+        post_id = post_json_object['object']['id']
+        post_id = remove_id_ending(post_id).replace('/', '--')
     for attach in post_json_object['object']['attachment']:
         if not (attach.get('mediaType') and attach.get('url')):
             continue
@@ -1126,7 +1132,25 @@ def get_post_attachments_as_html(post_json_object: {}, box_name: str,
             image_description = attach['name'].replace('"', "'")
         if _is_image_mime_type(media_type):
             image_url = attach['url']
-            if _is_attached_image(attach['url']) and 'svg' not in media_type:
+
+            # display svg images if they have first been rendered harmless
+            svg_harmless = True
+            if 'svg' in media_type:
+                svg_harmless = False
+                if '://' + domain_full + '/' in image_url:
+                    svg_harmless = True
+                else:
+                    if post_id:
+                        if '/' in image_url:
+                            im_filename = image_url.split('/')[-1]
+                        else:
+                            im_filename = image_url
+                        cached_svg_filename = \
+                            base_dir + '/media/' + post_id + '_' + im_filename
+                        if os.path.isfile(cached_svg_filename):
+                            svg_harmless = True
+
+            if _is_attached_image(attach['url']) and svg_harmless:
                 if not attachment_str:
                     attachment_str += '<div class="media">\n'
                     media_style_added = True