indymedia
/
epicyon
mirror of https://gitlab.com/bashrc2/epicyon
7
4
Fork 1
Code Issues 14 Releases Wiki Activity

Support for actor vcard

merge-requests/30/head
Bob Mottram 2022-02-15 14:42:00 +00:00
parent 1f20662c2b
commit 2023a5214a
2 changed files with 138 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
daemon.py
View File

@ -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,52 @@ class PubServer(BaseHTTPRequestHandler):
show_node_info_accounts,
referer_domain, debug, 5)
def _get_vcard(self, base_dir: str, path: str, calling_domain: str,
referer_domain: str, domain: str, debug: bool) -> bool:
if not self._has_accept(calling_domain):
return False
if 'text/vcard' not in self.headers['Accept']:
return False
if 'application/' in self.headers['Accept']:
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
if self.server.debug:
print('DEBUG: vcard ' + path)
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:
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', msglen,
None, calling_domain, True)
self._write(msg)
print('vcard sent to ' + str(referer_domain))
self.server.vcard_is_active = False
return True
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 +13653,11 @@ class PubServer(BaseHTTPRequestHandler):
fitness_performance(getreq_start_time, self.server.fitness,
'_GET', 'start', self.server.debug)
if self._get_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 +18982,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

114
pgp.py
View File

@ -21,6 +21,12 @@ 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 +341,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 +430,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 +626,53 @@ 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:' + 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;TYPE=PGP:' + pgp_key + '\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:
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'