2020-04-03 18:52:18 +00:00
|
|
|
__filename__ = "pgp.py"
|
|
|
|
__author__ = "Bob Mottram"
|
|
|
|
__license__ = "AGPL3+"
|
2021-01-26 10:07:42 +00:00
|
|
|
__version__ = "1.2.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"
|
2019-12-17 20:44:18 +00:00
|
|
|
|
2021-03-11 20:33:45 +00:00
|
|
|
import os
|
2021-03-11 17:15:32 +00:00
|
|
|
import subprocess
|
2021-03-11 20:33:45 +00:00
|
|
|
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
|
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
|
2021-03-11 17:15:32 +00:00
|
|
|
|
2019-12-17 20:44:18 +00:00
|
|
|
|
2021-12-28 17:33:54 +00:00
|
|
|
def get_email_address(actor_json: {}) -> str:
|
2019-12-17 20:44:18 +00:00
|
|
|
"""Returns the email address for the given actor
|
|
|
|
"""
|
2021-12-26 10:29:52 +00:00
|
|
|
if not actor_json.get('attachment'):
|
2019-12-17 20:44:18 +00:00
|
|
|
return ''
|
2021-12-26 10:32:45 +00:00
|
|
|
for property_value in actor_json['attachment']:
|
|
|
|
if not property_value.get('name'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value['name'].lower().startswith('email'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value.get('type'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value.get('value'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if property_value['type'] != 'PropertyValue':
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if '@' not in property_value['value']:
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if '.' not in property_value['value']:
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
return property_value['value']
|
2019-12-17 20:44:18 +00:00
|
|
|
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:
|
2019-12-17 20:44:18 +00:00
|
|
|
"""Returns PGP public key for the given actor
|
|
|
|
"""
|
2021-12-26 10:29:52 +00:00
|
|
|
if not actor_json.get('attachment'):
|
2019-12-17 20:44:18 +00:00
|
|
|
return ''
|
2021-12-26 10:32:45 +00:00
|
|
|
for property_value in actor_json['attachment']:
|
|
|
|
if not property_value.get('name'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value['name'].lower().startswith('pgp'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value.get('type'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value.get('value'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if property_value['type'] != 'PropertyValue':
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 19:15:36 +00:00
|
|
|
if not contains_pgp_public_key(property_value['value']):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
return property_value['value']
|
2019-12-17 20:44:18 +00:00
|
|
|
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']:
|
|
|
|
if not property_value.get('name'):
|
2020-07-06 09:52:06 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value['name'].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
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value.get('value'):
|
2020-07-06 09:52:06 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if property_value['type'] != 'PropertyValue':
|
2020-07-06 09:52:06 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if len(property_value['value']) < 10:
|
2020-07-06 09:52:06 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
return property_value['value']
|
2020-07-06 09:52:06 +00:00
|
|
|
return ''
|
|
|
|
|
|
|
|
|
2021-12-31 18:11:17 +00:00
|
|
|
def set_email_address(actor_json: {}, email_address: str) -> None:
|
2019-12-17 20:44:18 +00:00
|
|
|
"""Sets the email address for the given actor
|
|
|
|
"""
|
2020-07-06 10:25:18 +00:00
|
|
|
notEmailAddress = False
|
2021-12-31 18:11:17 +00:00
|
|
|
if '@' not in email_address:
|
2020-07-06 10:25:18 +00:00
|
|
|
notEmailAddress = True
|
2021-12-31 18:11:17 +00:00
|
|
|
if '.' not in email_address:
|
2020-07-06 10:25:18 +00:00
|
|
|
notEmailAddress = True
|
2021-12-31 18:11:17 +00:00
|
|
|
if '<' in email_address:
|
2020-12-12 15:31:28 +00:00
|
|
|
notEmailAddress = True
|
2021-12-31 18:11:17 +00:00
|
|
|
if email_address.startswith('@'):
|
2020-07-06 10:25:18 +00:00
|
|
|
notEmailAddress = True
|
|
|
|
|
2021-12-26 10:29:52 +00:00
|
|
|
if not actor_json.get('attachment'):
|
|
|
|
actor_json['attachment'] = []
|
2019-12-17 20:44:18 +00:00
|
|
|
|
2019-12-17 23:35:59 +00:00
|
|
|
# remove any existing value
|
2020-04-03 18:52:18 +00:00
|
|
|
propertyFound = None
|
2021-12-26 10:32:45 +00:00
|
|
|
for property_value in actor_json['attachment']:
|
|
|
|
if not property_value.get('name'):
|
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
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value['name'].lower().startswith('email'):
|
2019-12-17 23:35:59 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
propertyFound = property_value
|
2019-12-17 23:35:59 +00:00
|
|
|
break
|
|
|
|
if propertyFound:
|
2021-12-26 10:29:52 +00:00
|
|
|
actor_json['attachment'].remove(propertyFound)
|
2020-07-06 10:25:18 +00:00
|
|
|
if notEmailAddress:
|
2019-12-17 20:44:18 +00:00
|
|
|
return
|
|
|
|
|
2021-12-26 10:32:45 +00:00
|
|
|
for property_value in actor_json['attachment']:
|
|
|
|
if not property_value.get('name'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value.get('type'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value['name'].lower().startswith('email'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if property_value['type'] != 'PropertyValue':
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-31 18:11:17 +00:00
|
|
|
property_value['value'] = email_address
|
2019-12-17 20:44:18 +00:00
|
|
|
return
|
|
|
|
|
2020-04-03 18:52:18 +00:00
|
|
|
newEmailAddress = {
|
2019-12-17 20:44:18 +00:00
|
|
|
"name": "Email",
|
|
|
|
"type": "PropertyValue",
|
2021-12-31 18:11:17 +00:00
|
|
|
"value": email_address
|
2019-12-17 20:44:18 +00:00
|
|
|
}
|
2021-12-26 10:29:52 +00:00
|
|
|
actor_json['attachment'].append(newEmailAddress)
|
2019-12-17 20:44:18 +00:00
|
|
|
|
2020-04-03 18:52:18 +00:00
|
|
|
|
2021-12-31 18:11:17 +00:00
|
|
|
def set_pgp_pub_key(actor_json: {}, pgp_pub_key: str) -> None:
|
2019-12-17 20:44:18 +00:00
|
|
|
"""Sets a PGP public key for the given actor
|
|
|
|
"""
|
2020-07-06 09:52:06 +00:00
|
|
|
removeKey = False
|
2021-12-31 18:11:17 +00:00
|
|
|
if not pgp_pub_key:
|
2020-07-06 09:52:06 +00:00
|
|
|
removeKey = True
|
|
|
|
else:
|
2021-12-31 18:11:17 +00:00
|
|
|
if not contains_pgp_public_key(pgp_pub_key):
|
2020-07-06 09:52:06 +00:00
|
|
|
removeKey = True
|
2021-12-31 18:11:17 +00:00
|
|
|
if '<' in pgp_pub_key:
|
2020-12-12 15:31:28 +00:00
|
|
|
removeKey = 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 20:44:18 +00:00
|
|
|
|
2019-12-17 23:35:59 +00:00
|
|
|
# remove any existing value
|
2020-04-03 18:52:18 +00:00
|
|
|
propertyFound = None
|
2021-12-26 10:32:45 +00:00
|
|
|
for property_value in actor_json['attachment']:
|
|
|
|
if not property_value.get('name'):
|
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
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value['name'].lower().startswith('pgp'):
|
2019-12-17 23:35:59 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
propertyFound = property_value
|
2019-12-17 23:35:59 +00:00
|
|
|
break
|
|
|
|
if propertyFound:
|
2021-12-26 10:32:45 +00:00
|
|
|
actor_json['attachment'].remove(property_value)
|
2020-07-06 09:52:06 +00:00
|
|
|
if removeKey:
|
2019-12-17 20:44:18 +00:00
|
|
|
return
|
|
|
|
|
2021-12-26 10:32:45 +00:00
|
|
|
for property_value in actor_json['attachment']:
|
|
|
|
if not property_value.get('name'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value.get('type'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value['name'].lower().startswith('pgp'):
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if property_value['type'] != 'PropertyValue':
|
2019-12-17 20:44:18 +00:00
|
|
|
continue
|
2021-12-31 18:11:17 +00:00
|
|
|
property_value['value'] = pgp_pub_key
|
2019-12-17 20:44:18 +00:00
|
|
|
return
|
|
|
|
|
2021-12-31 18:11:17 +00:00
|
|
|
newpgp_pub_key = {
|
2019-12-17 20:44:18 +00:00
|
|
|
"name": "PGP",
|
|
|
|
"type": "PropertyValue",
|
2021-12-31 18:11:17 +00:00
|
|
|
"value": pgp_pub_key
|
2019-12-17 20:44:18 +00:00
|
|
|
}
|
2021-12-31 18:11:17 +00:00
|
|
|
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
|
|
|
|
"""
|
|
|
|
removeFingerprint = False
|
|
|
|
if not fingerprint:
|
|
|
|
removeFingerprint = True
|
|
|
|
else:
|
|
|
|
if len(fingerprint) < 10:
|
|
|
|
removeFingerprint = True
|
|
|
|
|
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
|
|
|
|
propertyFound = None
|
2021-12-26 10:32:45 +00:00
|
|
|
for property_value in actor_json['attachment']:
|
|
|
|
if not property_value.get('name'):
|
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
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value['name'].lower().startswith('openpgp'):
|
2020-07-06 09:52:06 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
propertyFound = property_value
|
2020-07-06 09:52:06 +00:00
|
|
|
break
|
|
|
|
if propertyFound:
|
2021-12-26 10:32:45 +00:00
|
|
|
actor_json['attachment'].remove(property_value)
|
2020-07-06 09:52:06 +00:00
|
|
|
if removeFingerprint:
|
|
|
|
return
|
|
|
|
|
2021-12-26 10:32:45 +00:00
|
|
|
for property_value in actor_json['attachment']:
|
|
|
|
if not property_value.get('name'):
|
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
|
2021-12-26 10:32:45 +00:00
|
|
|
if not property_value['name'].lower().startswith('openpgp'):
|
2020-07-06 09:52:06 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
if property_value['type'] != 'PropertyValue':
|
2020-07-06 09:52:06 +00:00
|
|
|
continue
|
2021-12-26 10:32:45 +00:00
|
|
|
property_value['value'] = fingerprint.strip()
|
2020-07-06 09:52:06 +00:00
|
|
|
return
|
|
|
|
|
2021-12-31 18:11:17 +00:00
|
|
|
newpgp_fingerprint = {
|
2020-07-06 09:52:06 +00:00
|
|
|
"name": "OpenPGP",
|
|
|
|
"type": "PropertyValue",
|
|
|
|
"value": fingerprint
|
|
|
|
}
|
2021-12-31 18:11:17 +00:00
|
|
|
actor_json['attachment'].append(newpgp_fingerprint)
|
2021-03-11 14:23:03 +00:00
|
|
|
|
|
|
|
|
2021-12-29 21:55:09 +00:00
|
|
|
def extract_pgp_public_key(content: str) -> str:
|
2021-03-11 14:23:03 +00:00
|
|
|
"""Returns the PGP key from the given text
|
|
|
|
"""
|
|
|
|
startBlock = '--BEGIN PGP PUBLIC KEY BLOCK--'
|
|
|
|
endBlock = '--END PGP PUBLIC KEY BLOCK--'
|
2021-03-11 17:15:32 +00:00
|
|
|
if startBlock not in content:
|
2021-03-11 14:23:03 +00:00
|
|
|
return None
|
2021-03-11 17:15:32 +00:00
|
|
|
if endBlock not in content:
|
2021-03-11 14:23:03 +00:00
|
|
|
return None
|
|
|
|
if '\n' not in content:
|
|
|
|
return None
|
|
|
|
linesList = content.split('\n')
|
|
|
|
extracting = False
|
|
|
|
publicKey = ''
|
|
|
|
for line in linesList:
|
|
|
|
if not extracting:
|
|
|
|
if startBlock in line:
|
|
|
|
extracting = True
|
|
|
|
else:
|
|
|
|
if endBlock in line:
|
|
|
|
publicKey += line
|
|
|
|
break
|
|
|
|
if extracting:
|
|
|
|
publicKey += line + '\n'
|
|
|
|
return publicKey
|
2021-03-11 17:15:32 +00:00
|
|
|
|
|
|
|
|
2021-12-29 21:55:09 +00:00
|
|
|
def _pgp_import_pub_key(recipientPubKey: str) -> str:
|
2021-03-11 17:15:32 +00:00
|
|
|
""" Import the given public key
|
|
|
|
"""
|
|
|
|
# do a dry run
|
|
|
|
cmdImportPubKey = \
|
|
|
|
'echo "' + recipientPubKey + '" | gpg --dry-run --import 2> /dev/null'
|
|
|
|
proc = subprocess.Popen([cmdImportPubKey],
|
|
|
|
stdout=subprocess.PIPE, shell=True)
|
|
|
|
(importResult, err) = proc.communicate()
|
|
|
|
if err:
|
|
|
|
return None
|
|
|
|
|
|
|
|
# this time for real
|
|
|
|
cmdImportPubKey = \
|
|
|
|
'echo "' + recipientPubKey + '" | gpg --import 2> /dev/null'
|
|
|
|
proc = subprocess.Popen([cmdImportPubKey],
|
|
|
|
stdout=subprocess.PIPE, shell=True)
|
|
|
|
(importResult, err) = proc.communicate()
|
|
|
|
if err:
|
|
|
|
return None
|
|
|
|
|
|
|
|
# get the key id
|
|
|
|
cmdImportPubKey = \
|
|
|
|
'echo "' + recipientPubKey + '" | gpg --show-keys'
|
|
|
|
proc = subprocess.Popen([cmdImportPubKey],
|
|
|
|
stdout=subprocess.PIPE, shell=True)
|
|
|
|
(importResult, err) = proc.communicate()
|
|
|
|
if not importResult:
|
|
|
|
return None
|
|
|
|
importResult = importResult.decode('utf-8').split('\n')
|
|
|
|
keyId = ''
|
|
|
|
for line in importResult:
|
|
|
|
if line.startswith('pub'):
|
|
|
|
continue
|
|
|
|
elif line.startswith('uid'):
|
|
|
|
continue
|
|
|
|
elif line.startswith('sub'):
|
|
|
|
continue
|
|
|
|
keyId = line.strip()
|
|
|
|
break
|
|
|
|
return keyId
|
|
|
|
|
|
|
|
|
2021-12-29 21:55:09 +00:00
|
|
|
def _pgp_encrypt(content: str, recipientPubKey: str) -> str:
|
2021-03-11 17:15:32 +00:00
|
|
|
""" Encrypt using your default pgp key to the given recipient
|
|
|
|
"""
|
2021-12-29 21:55:09 +00:00
|
|
|
keyId = _pgp_import_pub_key(recipientPubKey)
|
2021-03-11 17:15:32 +00:00
|
|
|
if not keyId:
|
|
|
|
return None
|
|
|
|
|
|
|
|
cmdEncrypt = \
|
|
|
|
'echo "' + content + '" | gpg --encrypt --armor --recipient ' + \
|
|
|
|
keyId + ' 2> /dev/null'
|
|
|
|
proc = subprocess.Popen([cmdEncrypt],
|
|
|
|
stdout=subprocess.PIPE, shell=True)
|
|
|
|
(encryptResult, err) = proc.communicate()
|
|
|
|
if not encryptResult:
|
|
|
|
return None
|
|
|
|
encryptResult = encryptResult.decode('utf-8')
|
2021-12-26 19:15:36 +00:00
|
|
|
if not is_pgp_encrypted(encryptResult):
|
2021-03-11 17:15:32 +00:00
|
|
|
return None
|
|
|
|
return encryptResult
|
|
|
|
|
|
|
|
|
2021-12-29 21:55:09 +00:00
|
|
|
def _get_pgp_public_key_from_actor(signing_priv_key_pem: str,
|
|
|
|
domain: str, handle: str,
|
|
|
|
actor_json: {} = None) -> str:
|
2021-03-11 19:13:41 +00:00
|
|
|
"""Searches tags on the actor to see if there is any PGP
|
|
|
|
public key specified
|
|
|
|
"""
|
2021-12-26 10:29:52 +00:00
|
|
|
if not actor_json:
|
|
|
|
actor_json, asHeader = \
|
2021-12-29 21:55:09 +00:00
|
|
|
get_actor_json(domain, handle, False, False, False, True,
|
|
|
|
signing_priv_key_pem, None)
|
2021-12-26 10:29:52 +00:00
|
|
|
if not actor_json:
|
2021-03-11 19:13:41 +00:00
|
|
|
return None
|
2021-12-26 10:29:52 +00:00
|
|
|
if not actor_json.get('attachment'):
|
2021-03-11 19:13:41 +00:00
|
|
|
return None
|
2021-12-26 10:29:52 +00:00
|
|
|
if not isinstance(actor_json['attachment'], list):
|
2021-03-11 19:13:41 +00:00
|
|
|
return None
|
2021-03-11 20:33:45 +00:00
|
|
|
# search through the tags on the actor
|
2021-12-26 10:29:52 +00:00
|
|
|
for tag in actor_json['attachment']:
|
2021-03-11 20:33:45 +00:00
|
|
|
if not isinstance(tag, dict):
|
|
|
|
continue
|
2021-03-11 19:13:41 +00:00
|
|
|
if not tag.get('value'):
|
|
|
|
continue
|
2021-03-11 20:35:48 +00:00
|
|
|
if not isinstance(tag['value'], str):
|
|
|
|
continue
|
2021-12-26 19:15:36 +00:00
|
|
|
if contains_pgp_public_key(tag['value']):
|
2021-03-12 09:50:08 +00:00
|
|
|
return tag['value']
|
2021-03-11 19:13:41 +00:00
|
|
|
return None
|
|
|
|
|
|
|
|
|
2021-12-29 21:55:09 +00:00
|
|
|
def has_local_pg_pkey() -> bool:
|
2021-03-11 20:33:45 +00:00
|
|
|
"""Returns true if there is a local .gnupg directory
|
|
|
|
"""
|
|
|
|
homeDir = str(Path.home())
|
|
|
|
gpgDir = homeDir + '/.gnupg'
|
2021-03-11 20:48:02 +00:00
|
|
|
if os.path.isdir(gpgDir):
|
2021-12-29 21:55:09 +00:00
|
|
|
keyId = pgp_local_public_key()
|
2021-05-05 09:31:35 +00:00
|
|
|
if keyId:
|
|
|
|
return True
|
2021-03-11 20:33:45 +00:00
|
|
|
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:
|
2021-03-11 20:33:45 +00:00
|
|
|
"""PGP encrypt a message to the given actor or handle
|
|
|
|
"""
|
|
|
|
# get the actor and extract the pgp public key from it
|
2021-08-31 14:17:11 +00:00
|
|
|
recipientPubKey = \
|
2021-12-29 21:55:09 +00:00
|
|
|
_get_pgp_public_key_from_actor(signing_priv_key_pem, domain, toHandle)
|
2021-03-11 20:33:45 +00:00
|
|
|
if not recipientPubKey:
|
|
|
|
return None
|
|
|
|
# encrypt using the recipient public key
|
2021-12-29 21:55:09 +00:00
|
|
|
return _pgp_encrypt(content, recipientPubKey)
|
2021-03-11 20:33:45 +00:00
|
|
|
|
|
|
|
|
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
|
2021-03-11 19:13:41 +00:00
|
|
|
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):
|
2021-12-29 21:55:09 +00:00
|
|
|
pubKey = extract_pgp_public_key(content)
|
2021-03-11 19:13:41 +00:00
|
|
|
else:
|
2021-08-31 14:17:11 +00:00
|
|
|
pubKey = \
|
2021-12-29 21:55:09 +00:00
|
|
|
_get_pgp_public_key_from_actor(signing_priv_key_pem,
|
|
|
|
domain, content, fromHandle)
|
2021-03-11 19:13:41 +00:00
|
|
|
if pubKey:
|
2021-12-29 21:55:09 +00:00
|
|
|
_pgp_import_pub_key(pubKey)
|
2021-03-11 17:15:32 +00:00
|
|
|
|
|
|
|
cmdDecrypt = \
|
|
|
|
'echo "' + content + '" | gpg --decrypt --armor 2> /dev/null'
|
|
|
|
proc = subprocess.Popen([cmdDecrypt],
|
|
|
|
stdout=subprocess.PIPE, shell=True)
|
|
|
|
(decryptResult, err) = proc.communicate()
|
|
|
|
if not decryptResult:
|
|
|
|
return content
|
2021-03-11 20:33:45 +00:00
|
|
|
decryptResult = decryptResult.decode('utf-8').strip()
|
2021-03-11 17:15:32 +00:00
|
|
|
return decryptResult
|
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
|
|
|
|
"""
|
|
|
|
cmdStr = \
|
|
|
|
"gpgconf --list-options gpg | " + \
|
|
|
|
"awk -F: '$1 == \"default-key\" {print $10}'"
|
|
|
|
proc = subprocess.Popen([cmdStr],
|
|
|
|
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
|
|
|
|
"""
|
2021-12-29 21:55:09 +00:00
|
|
|
keyId = _pgp_local_public_key_id()
|
2021-03-17 20:18:00 +00:00
|
|
|
if not keyId:
|
2021-05-05 10:13:26 +00:00
|
|
|
keyId = ''
|
2021-03-17 20:18:00 +00:00
|
|
|
cmdStr = "gpg --armor --export " + keyId
|
|
|
|
proc = subprocess.Popen([cmdStr],
|
|
|
|
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
|
|
|
|
|
|
|
|
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')
|
2021-12-31 18:11:17 +00:00
|
|
|
pgp_pub_key = pgp_local_public_key()
|
|
|
|
if not pgp_pub_key:
|
2021-03-17 20:18:00 +00:00
|
|
|
return None
|
2021-12-31 18:11:17 +00:00
|
|
|
pgp_pub_keyId = _pgp_local_public_key_id()
|
2021-03-17 20:18:00 +00:00
|
|
|
else:
|
|
|
|
if debug:
|
|
|
|
print('Testing with PGP public key ' + test)
|
2021-12-31 18:11:17 +00:00
|
|
|
pgp_pub_key = test
|
|
|
|
pgp_pub_keyId = 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)
|
|
|
|
|
2021-12-26 10:29:52 +00:00
|
|
|
actor_json, asHeader = \
|
2021-12-29 21:55:09 +00:00
|
|
|
get_actor_json(domain_full, handle, 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
|
2021-12-31 18:11:17 +00:00
|
|
|
if pgp_pub_keyId:
|
|
|
|
set_pgp_fingerprint(actor_json, pgp_pub_keyId)
|
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)
|
2021-12-31 18:11:17 +00:00
|
|
|
set_pgp_pub_key(actor_json, pgp_pub_key)
|
2021-03-17 20:18:00 +00:00
|
|
|
|
|
|
|
# create an actor update
|
2021-12-27 17:42:35 +00:00
|
|
|
statusNumber, published = get_status_number()
|
2021-03-17 20:18:00 +00:00
|
|
|
actorUpdate = {
|
|
|
|
'@context': 'https://www.w3.org/ns/activitystreams',
|
|
|
|
'id': actor + '#updates/' + statusNumber,
|
|
|
|
'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:
|
|
|
|
print('actor update is ' + str(actorUpdate))
|
|
|
|
|
|
|
|
# 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
|
|
|
|
|
|
|
|
postToBox = 'outbox'
|
|
|
|
|
|
|
|
# get the actor inbox for the To handle
|
2021-09-15 14:05:08 +00:00
|
|
|
originDomain = domain
|
2021-09-22 09:29:48 +00:00
|
|
|
(inboxUrl, pubKeyId, pubKey, fromPersonId, sharedInbox, avatarUrl,
|
2021-12-29 21:55:09 +00:00
|
|
|
displayName, _) = get_person_box(signing_priv_key_pem, originDomain,
|
2022-01-02 14:51:02 +00:00
|
|
|
base_dir, session, wf_request,
|
2021-12-29 21:55:09 +00:00
|
|
|
person_cache,
|
|
|
|
__version__, http_prefix, nickname,
|
|
|
|
domain, postToBox, 35725)
|
2021-03-17 20:18:00 +00:00
|
|
|
|
|
|
|
if not inboxUrl:
|
|
|
|
if debug:
|
|
|
|
print('DEBUG: No ' + postToBox + ' was found for ' + handle)
|
|
|
|
return None
|
|
|
|
if not fromPersonId:
|
|
|
|
if debug:
|
|
|
|
print('DEBUG: No actor was found for ' + handle)
|
|
|
|
return None
|
|
|
|
|
2021-12-28 21:36:27 +00:00
|
|
|
authHeader = create_basic_auth_header(nickname, password)
|
2021-03-17 20:18:00 +00:00
|
|
|
|
|
|
|
headers = {
|
|
|
|
'host': domain,
|
|
|
|
'Content-type': 'application/json',
|
|
|
|
'Authorization': authHeader
|
|
|
|
}
|
|
|
|
quiet = not debug
|
2021-03-17 21:23:52 +00:00
|
|
|
tries = 0
|
2021-03-17 20:18:00 +00:00
|
|
|
while tries < 4:
|
|
|
|
postResult = \
|
2021-12-29 21:55:09 +00:00
|
|
|
post_json(http_prefix, domain_full,
|
|
|
|
session, actorUpdate, [], inboxUrl,
|
|
|
|
headers, 5, quiet)
|
2021-03-17 20:18:00 +00:00
|
|
|
if postResult:
|
|
|
|
break
|
|
|
|
tries += 1
|
|
|
|
|
|
|
|
if postResult is None:
|
|
|
|
if debug:
|
|
|
|
print('DEBUG: POST pgp actor update failed for c2s to ' +
|
|
|
|
inboxUrl)
|
|
|
|
return None
|
|
|
|
|
|
|
|
if debug:
|
|
|
|
print('DEBUG: c2s POST pgp actor update success')
|
|
|
|
|
|
|
|
return actorUpdate
|