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

daemon.py

@@ -31,6 +31,7 @@ from metadata import meta_data_node_info
 from metadata import metadata_custom_emoji
 from enigma import get_enigma_pub_key
 from enigma import set_enigma_pub_key
+from pgp import actor_to_vcard
 from pgp import get_email_address
 from pgp import set_email_address
 from pgp import get_pgp_pub_key
@@ -1186,6 +1187,59 @@ class PubServer(BaseHTTPRequestHandler):
                                   show_node_info_accounts,
                                   referer_domain, debug, 5)
 
+    def _show_vcard(self, base_dir: str, path: str, calling_domain: str,
+                    referer_domain: str, domain: str, debug: bool) -> bool:
+        """Returns a vcard for the given account
+        """
+        if not self._has_accept(calling_domain):
+            return False
+        if path.endswith('.vcf'):
+            path = path.split('.vcf')[0]
+            accept_str = 'text/vcard'
+        else:
+            accept_str = self.headers['Accept']
+        if 'text/vcard' not in accept_str:
+            return False
+        if 'application/' in accept_str:
+            return False
+        if not path.startswith('/users/'):
+            self._400()
+            return True
+        nickname = path.split('/users/')[1]
+        if '/' in nickname:
+            nickname = nickname.split('/')[0]
+        if '?' in nickname:
+            nickname = nickname.split('?')[0]
+        if self.server.vcard_is_active:
+            print('vcard is busy during request from ' + str(referer_domain))
+            self._503()
+            return True
+        self.server.vcard_is_active = True
+        actor_json = None
+        actor_filename = \
+            acct_dir(base_dir, nickname, domain) + '.json'
+        if os.path.isfile(actor_filename):
+            actor_json = load_json(actor_filename)
+        if not actor_json:
+            print('WARN: vcard actor not found ' + actor_filename)
+            self._404()
+            self.server.vcard_is_active = False
+            return True
+        vcard_str = actor_to_vcard(actor_json)
+        if vcard_str:
+            msg = vcard_str.encode('utf-8')
+            msglen = len(msg)
+            self._set_headers('text/vcard; charset=utf-8', msglen,
+                              None, calling_domain, True)
+            self._write(msg)
+            print('vcard sent to ' + str(referer_domain))
+            self.server.vcard_is_active = False
+            return True
+        print('WARN: vcard string not returned')
+        self._404()
+        self.server.vcard_is_active = False
+        return True
+
     def _nodeinfo(self, ua_str: str, calling_domain: str,
                   referer_domain: str,
                   http_prefix: str, calling_site_timeout: int,
@@ -13606,6 +13660,11 @@ class PubServer(BaseHTTPRequestHandler):
         fitness_performance(getreq_start_time, self.server.fitness,
                             '_GET', 'start', self.server.debug)
 
+        if self._show_vcard(self.server.base_dir,
+                            self.path, calling_domain, referer_domain,
+                            self.server.domain, self.server.debug):
+            return
+
         # Since fediverse crawlers are quite active,
         # make returning info to them high priority
         # get nodeinfo endpoint
@@ -18930,6 +18989,7 @@ def run_daemon(dyslexic_font: bool,
     httpd.post_to_nickname = None
 
     httpd.nodeinfo_is_active = False
+    httpd.vcard_is_active = False
     httpd.masto_api_is_active = False
 
     httpd.dyslexic_font = dyslexic_font

epicyon-profile.css

@@ -194,6 +194,9 @@
     --voteresult-width: 80%;
     --voteresult-width-mobile: 80%;
     --voteresult-width-tiny: 40%;
+    --vcard-icon-size: 32px;
+    --vcard-icon-size-mobile: 80px;
+    --vcard-icon-size-tiny: 80px;
 }
 
 @font-face {
@@ -391,7 +394,6 @@ a:focus {
     color: #ffffff;
     text-align: var(--profile-text-align);
     line-height: 1.4em;
-    background-color: #141414;
 }
 
 .profileHeader .title {
@@ -1074,6 +1076,10 @@ div.container {
         font-size: var(--font-size);
         color: var(--title-color);
     }
+    .profileHeader img.vcard {
+        width: var(--vcard-icon-size);
+        float: right;
+    }
     .followText {
         font-size: var(--follow-text-size1);
         font-family: 'NimbusSanL';
@@ -1837,6 +1843,10 @@ div.container {
     blockquote {
         font-size: var(--quote-font-size-mobile);
     }
+    .profileHeader img.vcard {
+        width: var(--vcard-icon-size-mobile);
+        float: right;
+    }
     .accountsTable {
         width: 100%;
         border: 0;
@@ -2572,6 +2582,10 @@ div.container {
     blockquote {
         font-size: var(--quote-font-size-tiny);
     }
+    .profileHeader img.vcard {
+        width: var(--vcard-icon-size-tiny);
+        float: right;
+    }
     .accountsTable {
         width: 100%;
         border: 0;

epicyon.py

@@ -43,6 +43,7 @@ from posts import get_user_url
 from posts import check_domains
 from session import create_session
 from session import get_json
+from session import get_vcard
 from session import download_html
 from newswire import get_rss
 from filters import add_filter
@@ -288,6 +289,8 @@ parser.add_argument('--socnet', dest='socnet', type=str,
 parser.add_argument('--postsraw', dest='postsraw', type=str,
                     default=None,
                     help='Show raw json of posts for the given handle')
+parser.add_argument('--vcard', dest='vcard', type=str, default=None,
+                    help='Show the vcard for a given activitypub actor url')
 parser.add_argument('--json', dest='json', type=str, default=None,
                     help='Show the json for a given activitypub url')
 parser.add_argument('--htmlpost', dest='htmlpost', type=str, default=None,
@@ -967,6 +970,19 @@ if args.json:
         pprint(test_json)
     sys.exit()
 
+if args.vcard:
+    session = create_session(None)
+    if not args.domain:
+        args.domain = get_config_param(base_dir, 'domain')
+    domain = ''
+    if args.domain:
+        domain = args.domain
+    test_vcard = get_vcard(session, args.vcard,
+                           None, debug, __version__, http_prefix, domain)
+    if test_vcard:
+        print(test_vcard)
+    sys.exit()
+
 if args.htmlpost:
     session = create_session(None)
     profile_str = 'https://www.w3.org/ns/activitystreams'

pgp.py

@@ -8,6 +8,7 @@ __status__ = "Production"
 __module_group__ = "Profile Metadata"
 
 import os
+import base64
 import subprocess
 from pathlib import Path
 from person import get_actor_json
@@ -17,10 +18,17 @@ from utils import get_full_domain
 from utils import get_status_number
 from utils import local_actor_url
 from utils import replace_users_with_at
+from utils import remove_html
 from webfinger import webfinger_handle
 from posts import get_person_box
 from auth import create_basic_auth_header
 from session import post_json
+from xmpp import get_xmpp_address
+from jami import get_jami_address
+from matrix import get_matrix_address
+from briar import get_briar_address
+from cwtch import get_cwtch_address
+from blog import get_blog_address
 
 
 def get_email_address(actor_json: {}) -> str:
@@ -335,35 +343,6 @@ def _pgp_encrypt(content: str, recipient_pub_key: str) -> str:
     return encrypt_result
 
 
-def _get_pgp_public_key_from_actor(signing_priv_key_pem: str,
-                                   domain: str, handle: str,
-                                   actor_json: {} = None) -> str:
-    """Searches tags on the actor to see if there is any PGP
-    public key specified
-    """
-    if not actor_json:
-        actor_json, _ = \
-            get_actor_json(domain, handle, False, False, False, True,
-                           signing_priv_key_pem, None)
-    if not actor_json:
-        return None
-    if not actor_json.get('attachment'):
-        return None
-    if not isinstance(actor_json['attachment'], list):
-        return None
-    # search through the tags on the actor
-    for tag in actor_json['attachment']:
-        if not isinstance(tag, dict):
-            continue
-        if not tag.get('value'):
-            continue
-        if not isinstance(tag['value'], str):
-            continue
-        if contains_pgp_public_key(tag['value']):
-            return tag['value']
-    return None
-
-
 def has_local_pg_pkey() -> bool:
     """Returns true if there is a local .gnupg directory
     """
@@ -453,6 +432,35 @@ def pgp_local_public_key() -> str:
     return extract_pgp_public_key(result.decode('utf-8'))
 
 
+def _get_pgp_public_key_from_actor(signing_priv_key_pem: str,
+                                   domain: str, handle: str,
+                                   actor_json: {} = None) -> str:
+    """Searches tags on the actor to see if there is any PGP
+    public key specified
+    """
+    if not actor_json:
+        actor_json, _ = \
+            get_actor_json(domain, handle, False, False, False, True,
+                           signing_priv_key_pem, None)
+    if not actor_json:
+        return None
+    if not actor_json.get('attachment'):
+        return None
+    if not isinstance(actor_json['attachment'], list):
+        return None
+    # search through the tags on the actor
+    for tag in actor_json['attachment']:
+        if not isinstance(tag, dict):
+            continue
+        if not tag.get('value'):
+            continue
+        if not isinstance(tag['value'], str):
+            continue
+        if contains_pgp_public_key(tag['value']):
+            return tag['value']
+    return None
+
+
 def pgp_public_key_upload(base_dir: str, session,
                           nickname: str, password: str,
                           domain: str, port: int,
@@ -620,3 +628,57 @@ def pgp_public_key_upload(base_dir: str, session,
         print('DEBUG: c2s POST pgp actor update success')
 
     return actor_update
+
+
+def actor_to_vcard(actor: {}) -> str:
+    """Returns a vcard for a given actor
+    """
+    vcard_str = 'BEGIN:VCARD\n'
+    vcard_str += 'VERSION:4.0\n'
+    vcard_str += 'REV:' + actor['published'] + '\n'
+    vcard_str += 'FN:' + actor['name'] + '\n'
+    vcard_str += 'N:' + actor['preferredUsername'] + '\n'
+    vcard_str += 'URL:' + actor['url'] + '\n'
+    blog_address = get_blog_address(actor)
+    if blog_address:
+        vcard_str += 'URL:blog:' + blog_address + '\n'
+    vcard_str += 'NOTE:' + remove_html(actor['summary']) + '\n'
+    if actor['icon']['url']:
+        vcard_str += 'PHOTO:' + actor['icon']['url'] + '\n'
+    pgp_key = get_pgp_pub_key(actor)
+    if pgp_key:
+        vcard_str += 'KEY:data:application/pgp-keys;base64,' + \
+            base64.b64encode(pgp_key.encode('utf-8')).decode('utf-8') + '\n'
+    email_address = get_email_address(actor)
+    if email_address:
+        vcard_str += 'EMAIL;TYPE=internet:' + email_address + '\n'
+    xmpp_address = get_xmpp_address(actor)
+    if xmpp_address:
+        vcard_str += 'IMPP:xmpp:' + xmpp_address + '\n'
+    jami_address = get_jami_address(actor)
+    if jami_address:
+        vcard_str += 'IMPP:jami:' + jami_address + '\n'
+    matrix_address = get_matrix_address(actor)
+    if matrix_address:
+        vcard_str += 'IMPP:matrix:' + matrix_address + '\n'
+    briar_address = get_briar_address(actor)
+    if briar_address:
+        if briar_address.startswith('briar://'):
+            briar_address = briar_address.split('briar://')[1]
+        vcard_str += 'IMPP:briar:' + briar_address + '\n'
+    cwtch_address = get_cwtch_address(actor)
+    if cwtch_address:
+        vcard_str += 'IMPP:cwtch:' + cwtch_address + '\n'
+    if actor.get('hasOccupation'):
+        if len(actor['hasOccupation']) > 0:
+            if actor['hasOccupation'][0].get('name'):
+                vcard_str += \
+                    'ROLE:' + \
+                    actor['hasOccupation'][0]['name'] + '\n'
+            if actor['hasOccupation'][0].get('occupationLocation'):
+                city_name = \
+                    actor['hasOccupation'][0]['occupationLocation']['name']
+                vcard_str += \
+                    'ADR:;;;' + city_name + ';;;\n'
+    vcard_str += 'END:VCARD\n'
+    return vcard_str

session.py

@@ -246,6 +246,78 @@ def get_json(signing_priv_key_pem: str,
                              None, quiet, debug, True)
 
 
+def get_vcard(session, url: str, params: {}, debug: bool,
+              version: str = '1.3.0', http_prefix: str = 'https',
+              domain: str = 'testdomain',
+              timeout_sec: int = 20, quiet: bool = False) -> {}:
+    if not isinstance(url, str):
+        if debug and not quiet:
+            print('url: ' + str(url))
+            print('ERROR: get_vcard failed, url should be a string')
+        return None
+    headers = {
+        'Accept': 'text/vcard'
+    }
+    session_params = {}
+    session_headers = {}
+    if headers:
+        session_headers = headers
+    if params:
+        session_params = params
+    session_headers['User-Agent'] = 'Epicyon/' + version
+    if domain:
+        session_headers['User-Agent'] += \
+            '; +' + http_prefix + '://' + domain + '/'
+    if not session:
+        if not quiet:
+            print('WARN: get_vcard failed, no session specified for get_vcard')
+        return None
+
+    if debug:
+        HTTPConnection.debuglevel = 1
+
+    try:
+        result = session.get(url, headers=session_headers,
+                             params=session_params, timeout=timeout_sec)
+        if result.status_code != 200:
+            if result.status_code == 401:
+                print("WARN: get_vcard " + url + ' rejected by secure mode')
+            elif result.status_code == 403:
+                print('WARN: get_vcard Forbidden url: ' + url)
+            elif result.status_code == 404:
+                print('WARN: get_vcard Not Found url: ' + url)
+            elif result.status_code == 410:
+                print('WARN: get_vcard no longer available url: ' + url)
+            else:
+                print('WARN: get_vcard url: ' + url +
+                      ' failed with error code ' +
+                      str(result.status_code) +
+                      ' headers: ' + str(session_headers))
+        return result.content.decode('utf-8')
+    except requests.exceptions.RequestException as ex:
+        session_headers2 = session_headers.copy()
+        if session_headers2.get('Authorization'):
+            session_headers2['Authorization'] = 'REDACTED'
+        if debug and not quiet:
+            print('EX: get_vcard failed, url: ' + str(url) + ', ' +
+                  'headers: ' + str(session_headers2) + ', ' +
+                  'params: ' + str(session_params) + ', ' + str(ex))
+    except ValueError as ex:
+        session_headers2 = session_headers.copy()
+        if session_headers2.get('Authorization'):
+            session_headers2['Authorization'] = 'REDACTED'
+        if debug and not quiet:
+            print('EX: get_vcard failed, url: ' + str(url) + ', ' +
+                  'headers: ' + str(session_headers2) + ', ' +
+                  'params: ' + str(session_params) + ', ' + str(ex))
+    except SocketError as ex:
+        if not quiet:
+            if ex.errno == errno.ECONNRESET:
+                print('EX: get_vcard failed, ' +
+                      'connection was reset during get_vcard ' + str(ex))
+    return None
+
+
 def download_html(signing_priv_key_pem: str,
                   session, url: str, headers: {}, params: {}, debug: bool,
                   version: str = '1.3.0', http_prefix: str = 'https',

webapp_profile.py

@@ -453,6 +453,14 @@ def _get_profile_header(base_dir: str, http_prefix: str,
         '        <p>' + profile_description_short + '</p>\n' + login_button
     if pinned_content:
         html_str += pinned_content.replace('<p>', '<p>📎', 1)
+
+    # show vcard download link
+    html_str += \
+        '    <a href="/users/' + nickname + '.vcf" ' + \
+        'download="contact_' + nickname + '@' + domain_full + '.vcf">' + \
+        '<img class="vcard" src="/icons/vcard.png" ' + \
+        'title="vCard" alt="vCard" /></a>\n'
+
     html_str += \
         '      </figcaption>\n' + \
         '    </figure>\n\n'