epicyon/pgp.py

829 lines
27 KiB
Python
Raw Normal View History

2020-04-03 18:52:18 +00:00
__filename__ = "pgp.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
2022-02-03 13:58:20 +00:00
__version__ = "1.3.0"
2020-04-03 18:52:18 +00:00
__maintainer__ = "Bob Mottram"
2021-09-10 16:14:50 +00:00
__email__ = "bob@libreserver.org"
2020-04-03 18:52:18 +00:00
__status__ = "Production"
2021-06-15 15:08:12 +00:00
__module_group__ = "Profile Metadata"
import os
import base64
2021-03-11 17:15:32 +00:00
import subprocess
from pathlib import Path
2021-12-29 21:55:09 +00:00
from person import get_actor_json
2021-12-26 19:15:36 +00:00
from utils import contains_pgp_public_key
from utils import is_pgp_encrypted
2021-12-26 12:45:03 +00:00
from utils import get_full_domain
2021-12-27 17:42:35 +00:00
from utils import get_status_number
2021-12-26 10:19:59 +00:00
from utils import local_actor_url
2021-12-26 17:21:37 +00:00
from utils import replace_users_with_at
2022-02-15 16:15:57 +00:00
from utils import remove_html
2021-12-29 21:55:09 +00:00
from webfinger import webfinger_handle
from posts import get_person_box
2021-12-28 21:36:27 +00:00
from auth import create_basic_auth_header
2021-12-29 21:55:09 +00:00
from session import post_json
2022-02-15 14:42:00 +00:00
from xmpp import get_xmpp_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
2021-03-11 17:15:32 +00:00
from utils import get_attachment_property_value
2021-12-28 17:33:54 +00:00
def get_email_address(actor_json: {}) -> str:
"""Returns the email address for the given actor
"""
2021-12-26 10:29:52 +00:00
if not actor_json.get('attachment'):
return ''
2021-12-26 10:32:45 +00:00
for property_value in actor_json['attachment']:
2022-05-11 16:10:38 +00:00
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
continue
2022-05-11 16:10:38 +00:00
if not name_value.lower().startswith('email'):
continue
2021-12-26 10:32:45 +00:00
if not property_value.get('type'):
continue
prop_value_name, _ = \
get_attachment_property_value(property_value)
if not prop_value_name:
continue
2022-05-11 16:16:34 +00:00
if not property_value['type'].endswith('PropertyValue'):
continue
if '@' not in property_value[prop_value_name]:
continue
if '.' not in property_value[prop_value_name]:
continue
return property_value[prop_value_name]
return ''
2020-04-03 18:52:18 +00:00
2021-12-28 17:33:54 +00:00
def get_pgp_pub_key(actor_json: {}) -> str:
"""Returns PGP public key for the given actor
"""
2021-12-26 10:29:52 +00:00
if not actor_json.get('attachment'):
return ''
2021-12-26 10:32:45 +00:00
for property_value in actor_json['attachment']:
2022-05-11 16:10:38 +00:00
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
continue
2022-05-11 16:10:38 +00:00
if not name_value.lower().startswith('pgp'):
continue
2021-12-26 10:32:45 +00:00
if not property_value.get('type'):
continue
prop_value_name, _ = \
get_attachment_property_value(property_value)
if not prop_value_name:
continue
2022-05-11 16:16:34 +00:00
if not property_value['type'].endswith('PropertyValue'):
continue
if not contains_pgp_public_key(property_value[prop_value_name]):
continue
return property_value[prop_value_name]
return ''
2020-04-03 18:52:18 +00:00
2021-12-28 17:33:54 +00:00
def get_pgp_fingerprint(actor_json: {}) -> str:
2020-07-06 09:52:06 +00:00
"""Returns PGP fingerprint for the given actor
"""
2021-12-26 10:29:52 +00:00
if not actor_json.get('attachment'):
2020-07-06 09:52:06 +00:00
return ''
2021-12-26 10:32:45 +00:00
for property_value in actor_json['attachment']:
2022-05-11 16:10:38 +00:00
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
2020-07-06 09:52:06 +00:00
continue
2022-05-11 16:10:38 +00:00
if not name_value.lower().startswith('openpgp'):
2020-07-06 09:52:06 +00:00
continue
2021-12-26 10:32:45 +00:00
if not property_value.get('type'):
2020-07-06 09:52:06 +00:00
continue
prop_value_name, _ = \
get_attachment_property_value(property_value)
if not prop_value_name:
2020-07-06 09:52:06 +00:00
continue
2022-05-11 16:16:34 +00:00
if not property_value['type'].endswith('PropertyValue'):
2020-07-06 09:52:06 +00:00
continue
if len(property_value[prop_value_name]) < 10:
2020-07-06 09:52:06 +00:00
continue
return property_value[prop_value_name]
2020-07-06 09:52:06 +00:00
return ''
def set_email_address(actor_json: {}, email_address: str) -> None:
"""Sets the email address for the given actor
"""
2022-01-03 15:06:00 +00:00
not_email_address = False
if '@' not in email_address:
2022-01-03 15:06:00 +00:00
not_email_address = True
if '.' not in email_address:
2022-01-03 15:06:00 +00:00
not_email_address = True
if '<' in email_address:
2022-01-03 15:06:00 +00:00
not_email_address = True
if email_address.startswith('@'):
2022-01-03 15:06:00 +00:00
not_email_address = True
2020-07-06 10:25:18 +00:00
2021-12-26 10:29:52 +00:00
if not actor_json.get('attachment'):
actor_json['attachment'] = []
2019-12-17 23:35:59 +00:00
# remove any existing value
2022-01-03 10:27:55 +00:00
property_found = None
2021-12-26 10:32:45 +00:00
for property_value in actor_json['attachment']:
2022-05-11 16:10:38 +00:00
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
2019-12-17 23:35:59 +00:00
continue
2021-12-26 10:32:45 +00:00
if not property_value.get('type'):
2019-12-17 23:35:59 +00:00
continue
2022-05-11 16:10:38 +00:00
if not name_value.lower().startswith('email'):
2019-12-17 23:35:59 +00:00
continue
2022-01-03 10:27:55 +00:00
property_found = property_value
2019-12-17 23:35:59 +00:00
break
2022-01-03 10:27:55 +00:00
if property_found:
actor_json['attachment'].remove(property_found)
2022-01-03 15:06:00 +00:00
if not_email_address:
return
2021-12-26 10:32:45 +00:00
for property_value in actor_json['attachment']:
2022-05-11 16:10:38 +00:00
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
continue
2021-12-26 10:32:45 +00:00
if not property_value.get('type'):
continue
2022-05-11 16:10:38 +00:00
if not name_value.lower().startswith('email'):
continue
2022-05-11 16:16:34 +00:00
if not property_value['type'].endswith('PropertyValue'):
continue
prop_value_name, _ = \
get_attachment_property_value(property_value)
if not prop_value_name:
continue
property_value[prop_value_name] = email_address
return
2022-01-03 15:06:00 +00:00
new_email_address = {
"name": "Email",
"type": "PropertyValue",
"value": email_address
}
2022-01-03 15:06:00 +00:00
actor_json['attachment'].append(new_email_address)
2020-04-03 18:52:18 +00:00
def set_pgp_pub_key(actor_json: {}, pgp_pub_key: str) -> None:
"""Sets a PGP public key for the given actor
"""
2022-01-03 15:06:00 +00:00
remove_key = False
if not pgp_pub_key:
2022-01-03 15:06:00 +00:00
remove_key = True
2020-07-06 09:52:06 +00:00
else:
if not contains_pgp_public_key(pgp_pub_key):
2022-01-03 15:06:00 +00:00
remove_key = True
if '<' in pgp_pub_key:
2022-01-03 15:06:00 +00:00
remove_key = True
2020-07-06 09:52:06 +00:00
2021-12-26 10:29:52 +00:00
if not actor_json.get('attachment'):
actor_json['attachment'] = []
2019-12-17 23:35:59 +00:00
# remove any existing value
2022-01-03 10:27:55 +00:00
property_found = None
2021-12-26 10:32:45 +00:00
for property_value in actor_json['attachment']:
2022-05-11 16:10:38 +00:00
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
2019-12-17 23:35:59 +00:00
continue
2021-12-26 10:32:45 +00:00
if not property_value.get('type'):
2019-12-17 23:35:59 +00:00
continue
2022-05-11 16:10:38 +00:00
if not name_value.lower().startswith('pgp'):
2019-12-17 23:35:59 +00:00
continue
2022-01-03 10:27:55 +00:00
property_found = property_value
2019-12-17 23:35:59 +00:00
break
2022-01-03 10:27:55 +00:00
if property_found:
2022-05-30 15:15:17 +00:00
actor_json['attachment'].remove(property_found)
2022-01-03 15:06:00 +00:00
if remove_key:
return
2021-12-26 10:32:45 +00:00
for property_value in actor_json['attachment']:
2022-05-11 16:10:38 +00:00
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
continue
2021-12-26 10:32:45 +00:00
if not property_value.get('type'):
continue
2022-05-11 16:10:38 +00:00
if not name_value.lower().startswith('pgp'):
continue
2022-05-11 16:16:34 +00:00
if not property_value['type'].endswith('PropertyValue'):
continue
prop_value_name, _ = \
get_attachment_property_value(property_value)
if not prop_value_name:
continue
property_value[prop_value_name] = pgp_pub_key
return
newpgp_pub_key = {
"name": "PGP",
"type": "PropertyValue",
"value": pgp_pub_key
}
actor_json['attachment'].append(newpgp_pub_key)
2020-07-06 09:52:06 +00:00
2021-12-28 17:33:54 +00:00
def set_pgp_fingerprint(actor_json: {}, fingerprint: str) -> None:
2020-07-06 09:52:06 +00:00
"""Sets a PGP fingerprint for the given actor
"""
2022-01-03 15:06:00 +00:00
remove_fingerprint = False
2020-07-06 09:52:06 +00:00
if not fingerprint:
2022-01-03 15:06:00 +00:00
remove_fingerprint = True
2020-07-06 09:52:06 +00:00
else:
if len(fingerprint) < 10:
2022-01-03 15:06:00 +00:00
remove_fingerprint = True
2020-07-06 09:52:06 +00:00
2021-12-26 10:29:52 +00:00
if not actor_json.get('attachment'):
actor_json['attachment'] = []
2020-07-06 09:52:06 +00:00
# remove any existing value
2022-01-03 10:27:55 +00:00
property_found = None
2021-12-26 10:32:45 +00:00
for property_value in actor_json['attachment']:
2022-05-11 16:10:38 +00:00
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
2020-07-06 09:52:06 +00:00
continue
2021-12-26 10:32:45 +00:00
if not property_value.get('type'):
2020-07-06 09:52:06 +00:00
continue
2022-05-11 16:10:38 +00:00
if not name_value.lower().startswith('openpgp'):
2020-07-06 09:52:06 +00:00
continue
2022-01-03 10:27:55 +00:00
property_found = property_value
2020-07-06 09:52:06 +00:00
break
2022-01-03 10:27:55 +00:00
if property_found:
2022-05-30 15:15:17 +00:00
actor_json['attachment'].remove(property_found)
2022-01-03 15:06:00 +00:00
if remove_fingerprint:
2020-07-06 09:52:06 +00:00
return
2021-12-26 10:32:45 +00:00
for property_value in actor_json['attachment']:
2022-05-11 16:10:38 +00:00
name_value = None
if property_value.get('name'):
name_value = property_value['name']
elif property_value.get('schema:name'):
name_value = property_value['schema:name']
if not name_value:
2020-07-06 09:52:06 +00:00
continue
2021-12-26 10:32:45 +00:00
if not property_value.get('type'):
2020-07-06 09:52:06 +00:00
continue
2022-05-11 16:10:38 +00:00
if not name_value.lower().startswith('openpgp'):
2020-07-06 09:52:06 +00:00
continue
2022-05-11 16:16:34 +00:00
if not property_value['type'].endswith('PropertyValue'):
2020-07-06 09:52:06 +00:00
continue
prop_value_name, _ = \
get_attachment_property_value(property_value)
if not prop_value_name:
continue
property_value[prop_value_name] = fingerprint.strip()
2020-07-06 09:52:06 +00:00
return
newpgp_fingerprint = {
2020-07-06 09:52:06 +00:00
"name": "OpenPGP",
"type": "PropertyValue",
"value": fingerprint
}
actor_json['attachment'].append(newpgp_fingerprint)
2021-12-29 21:55:09 +00:00
def extract_pgp_public_key(content: str) -> str:
"""Returns the PGP key from the given text
"""
2022-01-03 15:06:00 +00:00
start_block = '--BEGIN PGP PUBLIC KEY BLOCK--'
end_block = '--END PGP PUBLIC KEY BLOCK--'
if start_block not in content:
return None
2022-01-03 15:06:00 +00:00
if end_block not in content:
return None
if '\n' not in content:
return None
2022-01-03 15:06:00 +00:00
lines_list = content.split('\n')
extracting = False
2022-01-03 15:06:00 +00:00
public_key = ''
for line in lines_list:
if not extracting:
2022-01-03 15:06:00 +00:00
if start_block in line:
extracting = True
else:
2022-01-03 15:06:00 +00:00
if end_block in line:
public_key += line
break
if extracting:
2022-01-03 15:06:00 +00:00
public_key += line + '\n'
return public_key
2021-03-11 17:15:32 +00:00
2022-01-03 15:06:00 +00:00
def _pgp_import_pub_key(recipient_pub_key: str) -> str:
2021-03-11 17:15:32 +00:00
""" Import the given public key
"""
# do a dry run
2022-01-03 15:06:00 +00:00
cmd_import_pub_key = \
'echo "' + recipient_pub_key + \
'" | gpg --dry-run --import 2> /dev/null'
proc = subprocess.Popen([cmd_import_pub_key],
2021-03-11 17:15:32 +00:00
stdout=subprocess.PIPE, shell=True)
2022-01-03 15:06:00 +00:00
(import_result, err) = proc.communicate()
2021-03-11 17:15:32 +00:00
if err:
return None
# this time for real
2022-01-03 15:06:00 +00:00
cmd_import_pub_key = \
'echo "' + recipient_pub_key + '" | gpg --import 2> /dev/null'
proc = subprocess.Popen([cmd_import_pub_key],
2021-03-11 17:15:32 +00:00
stdout=subprocess.PIPE, shell=True)
2022-01-03 15:06:00 +00:00
(import_result, err) = proc.communicate()
2021-03-11 17:15:32 +00:00
if err:
return None
# get the key id
2022-01-03 15:06:00 +00:00
cmd_import_pub_key = \
'echo "' + recipient_pub_key + '" | gpg --show-keys'
proc = subprocess.Popen([cmd_import_pub_key],
2021-03-11 17:15:32 +00:00
stdout=subprocess.PIPE, shell=True)
2022-01-03 15:06:00 +00:00
(import_result, err) = proc.communicate()
if not import_result:
2021-03-11 17:15:32 +00:00
return None
2022-01-03 15:06:00 +00:00
import_result = import_result.decode('utf-8').split('\n')
key_id = ''
for line in import_result:
2021-03-11 17:15:32 +00:00
if line.startswith('pub'):
continue
2022-01-03 15:06:00 +00:00
if line.startswith('uid'):
2021-03-11 17:15:32 +00:00
continue
2022-01-03 15:06:00 +00:00
if line.startswith('sub'):
2021-03-11 17:15:32 +00:00
continue
2022-01-03 15:06:00 +00:00
key_id = line.strip()
2021-03-11 17:15:32 +00:00
break
2022-01-03 15:06:00 +00:00
return key_id
2021-03-11 17:15:32 +00:00
2022-01-03 15:06:00 +00:00
def _pgp_encrypt(content: str, recipient_pub_key: str) -> str:
2021-03-11 17:15:32 +00:00
""" Encrypt using your default pgp key to the given recipient
"""
2022-01-03 15:06:00 +00:00
key_id = _pgp_import_pub_key(recipient_pub_key)
if not key_id:
2021-03-11 17:15:32 +00:00
return None
2022-01-03 15:06:00 +00:00
cmd_encrypt = \
2021-03-11 17:15:32 +00:00
'echo "' + content + '" | gpg --encrypt --armor --recipient ' + \
2022-01-03 15:06:00 +00:00
key_id + ' 2> /dev/null'
proc = subprocess.Popen([cmd_encrypt],
2021-03-11 17:15:32 +00:00
stdout=subprocess.PIPE, shell=True)
2022-01-03 15:06:00 +00:00
(encrypt_result, _) = proc.communicate()
if not encrypt_result:
2021-03-11 17:15:32 +00:00
return None
2022-01-03 15:06:00 +00:00
encrypt_result = encrypt_result.decode('utf-8')
if not is_pgp_encrypted(encrypt_result):
2021-03-11 17:15:32 +00:00
return None
2022-01-03 15:06:00 +00:00
return encrypt_result
2021-03-11 17:15:32 +00:00
2021-12-29 21:55:09 +00:00
def has_local_pg_pkey() -> bool:
"""Returns true if there is a local .gnupg directory
"""
2022-01-03 15:06:00 +00:00
home_dir = str(Path.home())
gpg_dir = home_dir + '/.gnupg'
if os.path.isdir(gpg_dir):
key_id = pgp_local_public_key()
if key_id:
2021-05-05 09:31:35 +00:00
return True
return False
2021-12-29 21:55:09 +00:00
def pgp_encrypt_to_actor(domain: str, content: str, toHandle: str,
signing_priv_key_pem: str) -> str:
"""PGP encrypt a message to the given actor or handle
"""
# get the actor and extract the pgp public key from it
2022-01-03 15:06:00 +00:00
recipient_pub_key = \
2021-12-29 21:55:09 +00:00
_get_pgp_public_key_from_actor(signing_priv_key_pem, domain, toHandle)
2022-01-03 15:06:00 +00:00
if not recipient_pub_key:
return None
# encrypt using the recipient public key
2022-01-03 15:06:00 +00:00
return _pgp_encrypt(content, recipient_pub_key)
2021-12-29 21:55:09 +00:00
def pgp_decrypt(domain: str, content: str, fromHandle: str,
signing_priv_key_pem: str) -> str:
2021-03-11 17:15:32 +00:00
""" Encrypt using your default pgp key to the given recipient
fromHandle can be a handle or actor url
2021-03-11 17:15:32 +00:00
"""
2021-12-26 19:15:36 +00:00
if not is_pgp_encrypted(content):
2021-03-11 17:15:32 +00:00
return content
# if the public key is also included within the message then import it
2021-12-26 19:15:36 +00:00
if contains_pgp_public_key(content):
2022-01-03 15:06:00 +00:00
pub_key = extract_pgp_public_key(content)
else:
2022-01-03 15:06:00 +00:00
pub_key = \
2021-12-29 21:55:09 +00:00
_get_pgp_public_key_from_actor(signing_priv_key_pem,
domain, content, fromHandle)
2022-01-03 15:06:00 +00:00
if pub_key:
_pgp_import_pub_key(pub_key)
2021-03-11 17:15:32 +00:00
2022-01-03 15:06:00 +00:00
cmd_decrypt = \
2021-03-11 17:15:32 +00:00
'echo "' + content + '" | gpg --decrypt --armor 2> /dev/null'
2022-01-03 15:06:00 +00:00
proc = subprocess.Popen([cmd_decrypt],
2021-03-11 17:15:32 +00:00
stdout=subprocess.PIPE, shell=True)
2022-01-03 15:06:00 +00:00
(decrypt_result, _) = proc.communicate()
if not decrypt_result:
2021-03-11 17:15:32 +00:00
return content
2022-01-03 15:06:00 +00:00
decrypt_result = decrypt_result.decode('utf-8').strip()
return decrypt_result
2021-03-17 20:18:00 +00:00
2021-12-29 21:55:09 +00:00
def _pgp_local_public_key_id() -> str:
2021-03-17 20:18:00 +00:00
"""Gets the local pgp public key ID
"""
2022-01-03 15:06:00 +00:00
cmd_str = \
2021-03-17 20:18:00 +00:00
"gpgconf --list-options gpg | " + \
"awk -F: '$1 == \"default-key\" {print $10}'"
2022-01-03 15:06:00 +00:00
proc = subprocess.Popen([cmd_str],
2021-03-17 20:18:00 +00:00
stdout=subprocess.PIPE, shell=True)
(result, err) = proc.communicate()
if err:
return None
if not result:
return None
if len(result) < 5:
return None
2021-03-17 20:23:44 +00:00
return result.decode('utf-8').replace('"', '').strip()
2021-03-17 20:18:00 +00:00
2021-12-29 21:55:09 +00:00
def pgp_local_public_key() -> str:
2021-03-17 20:18:00 +00:00
"""Gets the local pgp public key
"""
2022-01-03 15:06:00 +00:00
key_id = _pgp_local_public_key_id()
if not key_id:
key_id = ''
cmd_str = "gpg --armor --export " + key_id
proc = subprocess.Popen([cmd_str],
2021-03-17 20:18:00 +00:00
stdout=subprocess.PIPE, shell=True)
(result, err) = proc.communicate()
if err:
return None
if not result:
return None
2021-12-29 21:55:09 +00:00
return extract_pgp_public_key(result.decode('utf-8'))
2021-03-17 20:18:00 +00:00
2022-02-15 14:42:00 +00:00
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, _ = \
2022-04-29 13:54:13 +00:00
get_actor_json(domain, handle, False, False, False, False,
False, True, signing_priv_key_pem, None)
2022-02-15 14:42:00 +00:00
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
prop_value_name, _ = get_attachment_property_value(tag)
if not prop_value_name:
2022-02-15 14:42:00 +00:00
continue
if not isinstance(tag[prop_value_name], str):
2022-02-15 14:42:00 +00:00
continue
if contains_pgp_public_key(tag[prop_value_name]):
return tag[prop_value_name]
2022-02-15 14:42:00 +00:00
return None
2021-12-29 21:55:09 +00:00
def pgp_public_key_upload(base_dir: str, session,
nickname: str, password: str,
domain: str, port: int,
http_prefix: str,
cached_webfingers: {}, person_cache: {},
debug: bool, test: str,
signing_priv_key_pem: str) -> {}:
2021-03-17 20:18:00 +00:00
if debug:
2021-12-29 21:55:09 +00:00
print('pgp_public_key_upload')
2021-03-17 20:18:00 +00:00
if not session:
if debug:
2021-12-29 21:55:09 +00:00
print('WARN: No session for pgp_public_key_upload')
2021-03-17 20:18:00 +00:00
return None
if not test:
if debug:
print('Getting PGP public key')
pgp_pub_key = pgp_local_public_key()
if not pgp_pub_key:
2021-03-17 20:18:00 +00:00
return None
2022-01-03 15:06:00 +00:00
pgp_pub_key_id = _pgp_local_public_key_id()
2021-03-17 20:18:00 +00:00
else:
if debug:
print('Testing with PGP public key ' + test)
pgp_pub_key = test
2022-01-03 15:06:00 +00:00
pgp_pub_key_id = None
2021-03-17 20:18:00 +00:00
2021-12-26 12:45:03 +00:00
domain_full = get_full_domain(domain, port)
2021-03-17 20:18:00 +00:00
if debug:
2021-12-26 10:00:46 +00:00
print('PGP test domain: ' + domain_full)
2021-03-17 20:18:00 +00:00
2021-12-26 10:00:46 +00:00
handle = nickname + '@' + domain_full
2021-03-17 20:18:00 +00:00
if debug:
print('Getting actor for ' + handle)
2022-01-03 15:06:00 +00:00
actor_json, _ = \
2022-04-29 13:54:13 +00:00
get_actor_json(domain_full, handle, False, False, False, False,
debug, True, signing_priv_key_pem, session)
2021-12-26 10:29:52 +00:00
if not actor_json:
2021-03-17 20:18:00 +00:00
if debug:
print('No actor returned for ' + handle)
return None
if debug:
print('Actor for ' + handle + ' obtained')
2021-12-26 10:19:59 +00:00
actor = local_actor_url(http_prefix, nickname, domain_full)
2021-12-26 17:21:37 +00:00
handle = replace_users_with_at(actor)
2021-03-17 20:18:00 +00:00
# check that this looks like the correct actor
2021-12-26 10:29:52 +00:00
if not actor_json.get('id'):
2021-03-17 20:18:00 +00:00
if debug:
print('Actor has no id')
return None
2021-12-26 10:29:52 +00:00
if not actor_json.get('url'):
2021-03-17 20:18:00 +00:00
if debug:
print('Actor has no url')
return None
2021-12-26 10:29:52 +00:00
if not actor_json.get('type'):
2021-03-17 20:18:00 +00:00
if debug:
print('Actor has no type')
return None
2021-12-26 10:29:52 +00:00
if actor_json['id'] != actor:
2021-03-17 20:18:00 +00:00
if debug:
print('Actor id is not ' + actor +
2021-12-26 10:29:52 +00:00
' instead is ' + actor_json['id'])
2021-03-17 20:18:00 +00:00
return None
2021-12-26 10:29:52 +00:00
if actor_json['url'] != handle:
2021-03-17 20:18:00 +00:00
if debug:
print('Actor url is not ' + handle)
return None
2021-12-26 10:29:52 +00:00
if actor_json['type'] != 'Person':
2021-03-17 20:18:00 +00:00
if debug:
print('Actor type is not Person')
return None
# set the pgp details
2022-01-03 15:06:00 +00:00
if pgp_pub_key_id:
set_pgp_fingerprint(actor_json, pgp_pub_key_id)
2021-03-17 20:18:00 +00:00
else:
if debug:
print('No PGP key Id. Continuing anyway.')
if debug:
print('Setting PGP key within ' + actor)
set_pgp_pub_key(actor_json, pgp_pub_key)
2021-03-17 20:18:00 +00:00
# create an actor update
2022-01-03 15:06:00 +00:00
status_number, _ = get_status_number()
actor_update = {
2021-03-17 20:18:00 +00:00
'@context': 'https://www.w3.org/ns/activitystreams',
2022-01-03 15:06:00 +00:00
'id': actor + '#updates/' + status_number,
2021-03-17 20:18:00 +00:00
'type': 'Update',
'actor': actor,
'to': [actor],
'cc': [],
2021-12-26 10:29:52 +00:00
'object': actor_json
2021-03-17 20:18:00 +00:00
}
if debug:
2022-01-03 15:06:00 +00:00
print('actor update is ' + str(actor_update))
2021-03-17 20:18:00 +00:00
# lookup the inbox for the To handle
2022-01-02 14:51:02 +00:00
wf_request = \
2021-12-29 21:55:09 +00:00
webfinger_handle(session, handle, http_prefix, cached_webfingers,
domain, __version__, debug, False,
signing_priv_key_pem)
2022-01-02 14:51:02 +00:00
if not wf_request:
2021-03-17 20:18:00 +00:00
if debug:
print('DEBUG: pgp actor update webfinger failed for ' +
handle)
return None
2022-01-02 14:51:02 +00:00
if not isinstance(wf_request, dict):
2021-03-17 20:18:00 +00:00
if debug:
print('WARN: Webfinger for ' + handle +
2022-01-02 14:51:02 +00:00
' did not return a dict. ' + str(wf_request))
2021-03-17 20:18:00 +00:00
return None
2022-01-03 15:06:00 +00:00
post_to_box = 'outbox'
2021-03-17 20:18:00 +00:00
# get the actor inbox for the To handle
2022-01-03 15:06:00 +00:00
origin_domain = domain
(inbox_url, _, _, from_person_id, _, _,
_, _) = get_person_box(signing_priv_key_pem, origin_domain,
base_dir, session, wf_request,
person_cache,
__version__, http_prefix, nickname,
domain, post_to_box, 35725)
if not inbox_url:
2021-03-17 20:18:00 +00:00
if debug:
2022-01-03 15:06:00 +00:00
print('DEBUG: No ' + post_to_box + ' was found for ' + handle)
2021-03-17 20:18:00 +00:00
return None
2022-01-03 15:06:00 +00:00
if not from_person_id:
2021-03-17 20:18:00 +00:00
if debug:
print('DEBUG: No actor was found for ' + handle)
return None
2022-01-03 15:06:00 +00:00
auth_header = create_basic_auth_header(nickname, password)
2021-03-17 20:18:00 +00:00
headers = {
'host': domain,
'Content-type': 'application/json',
2022-01-03 15:06:00 +00:00
'Authorization': auth_header
2021-03-17 20:18:00 +00:00
}
quiet = not debug
2021-03-17 21:23:52 +00:00
tries = 0
2021-03-17 20:18:00 +00:00
while tries < 4:
2022-01-03 15:06:00 +00:00
post_result = \
2021-12-29 21:55:09 +00:00
post_json(http_prefix, domain_full,
2022-01-03 15:06:00 +00:00
session, actor_update, [], inbox_url,
2021-12-29 21:55:09 +00:00
headers, 5, quiet)
2022-01-03 15:06:00 +00:00
if post_result:
2021-03-17 20:18:00 +00:00
break
tries += 1
2022-01-03 15:06:00 +00:00
if post_result is None:
2021-03-17 20:18:00 +00:00
if debug:
print('DEBUG: POST pgp actor update failed for c2s to ' +
2022-01-03 15:06:00 +00:00
inbox_url)
2021-03-17 20:18:00 +00:00
return None
if debug:
print('DEBUG: c2s POST pgp actor update success')
2022-01-03 15:06:00 +00:00
return actor_update
2022-02-15 14:42:00 +00:00
2022-02-16 11:50:35 +00:00
def actor_to_vcard(actor: {}, domain: str) -> str:
2022-02-15 14:42:00 +00:00
"""Returns a vcard for a given actor
"""
vcard_str = 'BEGIN:VCARD\n'
vcard_str += 'VERSION:4.0\n'
vcard_str += 'REV:' + actor['published'] + '\n'
2022-02-16 11:29:40 +00:00
vcard_str += 'FN:' + remove_html(actor['name']) + '\n'
2022-02-16 11:10:44 +00:00
vcard_str += 'NICKNAME:' + actor['preferredUsername'] + '\n'
2022-02-16 11:50:35 +00:00
vcard_str += 'URL;TYPE=profile:' + actor['url'] + '\n'
2022-02-15 14:42:00 +00:00
blog_address = get_blog_address(actor)
if blog_address:
2022-02-16 11:50:35 +00:00
vcard_str += 'URL;TYPE=blog:' + blog_address + '\n'
2022-02-15 16:15:57 +00:00
vcard_str += 'NOTE:' + remove_html(actor['summary']) + '\n'
2022-02-15 14:42:00 +00:00
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,' + \
2022-02-15 16:11:22 +00:00
base64.b64encode(pgp_key.encode('utf-8')).decode('utf-8') + '\n'
2022-02-15 14:42:00 +00:00
email_address = get_email_address(actor)
if email_address:
vcard_str += 'EMAIL;TYPE=internet:' + email_address + '\n'
2022-02-16 12:38:06 +00:00
vcard_str += 'IMPP:fediverse:' + \
2022-02-16 11:50:35 +00:00
actor['preferredUsername'] + '@' + domain + '\n'
2022-02-15 14:42:00 +00:00
xmpp_address = get_xmpp_address(actor)
if xmpp_address:
2022-02-16 12:38:06 +00:00
vcard_str += 'IMPP:xmpp:' + xmpp_address + '\n'
2022-02-15 14:42:00 +00:00
matrix_address = get_matrix_address(actor)
if matrix_address:
2022-02-16 12:38:06 +00:00
vcard_str += 'IMPP:matrix:' + matrix_address + '\n'
2022-02-15 14:42:00 +00:00
briar_address = get_briar_address(actor)
if briar_address:
2022-02-15 18:10:36 +00:00
if briar_address.startswith('briar://'):
briar_address = briar_address.split('briar://')[1]
2022-02-16 12:38:06 +00:00
vcard_str += 'IMPP:briar:' + briar_address + '\n'
2022-02-15 14:42:00 +00:00
cwtch_address = get_cwtch_address(actor)
if cwtch_address:
2022-02-16 12:38:06 +00:00
vcard_str += 'IMPP:cwtch:' + cwtch_address + '\n'
2022-02-15 14:42:00 +00:00
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'
2022-02-15 15:48:57 +00:00
return vcard_str
2022-02-16 11:10:44 +00:00
2022-02-16 11:50:35 +00:00
def actor_to_vcard_xml(actor: {}, domain: str) -> str:
2022-02-16 11:10:44 +00:00
"""Returns a xml formatted vcard for a given actor
"""
vcard_str = '<?xml version="1.0" encoding="UTF-8"?>\n'
vcard_str += '<vcards xmlns="urn:ietf:params:xml:ns:vcard-4.0">\n'
vcard_str += ' <vcard>\n'
2022-02-16 11:29:40 +00:00
vcard_str += ' <fn><text>' + \
remove_html(actor['name']) + '</text></fn>\n'
2022-02-16 11:10:44 +00:00
vcard_str += ' <nickname><text>' + \
actor['preferredUsername'] + '</text></nickname>\n'
2022-02-16 11:16:15 +00:00
vcard_str += ' <note><text>' + \
remove_html(actor['summary']) + '</text></note>\n'
2022-02-16 11:10:44 +00:00
email_address = get_email_address(actor)
if email_address:
vcard_str += ' <email><text>' + email_address + '</text></email>\n'
2022-02-16 11:50:35 +00:00
vcard_str += ' <impp>' + \
'<parameters><type><text>fediverse</text></type></parameters>' + \
'<text>' + actor['preferredUsername'] + '@' + domain + \
'</text></impp>\n'
2022-02-16 11:10:44 +00:00
xmpp_address = get_xmpp_address(actor)
if xmpp_address:
vcard_str += ' <impp>' + \
'<parameters><type><text>xmpp</text></type></parameters>' + \
'<text>' + xmpp_address + '</text></impp>\n'
matrix_address = get_matrix_address(actor)
if matrix_address:
vcard_str += ' <impp>' + \
'<parameters><type><text>matrix</text></type></parameters>' + \
'<text>' + matrix_address + '</text></impp>\n'
briar_address = get_briar_address(actor)
if briar_address:
vcard_str += ' <impp>' + \
'<parameters><type><text>briar</text></type></parameters>' + \
'<uri>' + briar_address + '</uri></impp>\n'
cwtch_address = get_cwtch_address(actor)
if cwtch_address:
vcard_str += ' <impp>' + \
'<parameters><type><text>cwtch</text></type></parameters>' + \
'<text>' + cwtch_address + '</text></impp>\n'
vcard_str += ' <url>' + \
'<parameters><type><text>profile</text></type></parameters>' + \
'<uri>' + actor['url'] + '</uri></url>\n'
blog_address = get_blog_address(actor)
if blog_address:
vcard_str += ' <url>' + \
'<parameters><type><text>blog</text></type></parameters>' + \
'<uri>' + blog_address + '</uri></url>\n'
vcard_str += ' <rev>' + actor['published'] + '</rev>\n'
if actor['icon']['url']:
vcard_str += \
' <photo><uri>' + actor['icon']['url'] + '</uri></photo>\n'
pgp_key = get_pgp_pub_key(actor)
if pgp_key:
pgp_key_encoded = \
base64.b64encode(pgp_key.encode('utf-8')).decode('utf-8')
vcard_str += \
' <key>' + \
'<parameters>' + \
'<type><text>data</text></type>' + \
'<mediatype>application/pgp-keys;base64</mediatype>' + \
'</parameters>' + \
'<text>' + pgp_key_encoded + '</text></key>\n'
if actor.get('hasOccupation'):
if len(actor['hasOccupation']) > 0:
if actor['hasOccupation'][0].get('name'):
vcard_str += \
' <role><text>' + \
actor['hasOccupation'][0]['name'] + '</text></role>\n'
if actor['hasOccupation'][0].get('occupationLocation'):
city_name = \
actor['hasOccupation'][0]['occupationLocation']['name']
vcard_str += \
' <adr><locality>' + city_name + '</locality></adr>\n'
vcard_str += ' </vcard>\n'
vcard_str += '</vcards>\n'
return vcard_str