mirror of https://gitlab.com/bashrc2/epicyon
				
				
				
			Decrypt pgp encrypted DMs
							parent
							
								
									09a7b3200b
								
							
						
					
					
						commit
						b7c1b02346
					
				
							
								
								
									
										18
									
								
								content.py
								
								
								
								
							
							
						
						
									
										18
									
								
								content.py
								
								
								
								
							| 
						 | 
				
			
			@ -65,6 +65,9 @@ def _removeQuotesWithinQuotes(content: str) -> str:
 | 
			
		|||
def htmlReplaceEmailQuote(content: str) -> str:
 | 
			
		||||
    """Replaces an email style quote "> Some quote" with html blockquote
 | 
			
		||||
    """
 | 
			
		||||
    if '--BEGIN PGP MESSAGE--' in content or \
 | 
			
		||||
       '--BEGIN PGP PUBLIC KEY BLOCK--' in content:
 | 
			
		||||
        return content
 | 
			
		||||
    # replace quote paragraph
 | 
			
		||||
    if '<p>"' in content:
 | 
			
		||||
        if '"</p>' in content:
 | 
			
		||||
| 
						 | 
				
			
			@ -106,6 +109,9 @@ def htmlReplaceQuoteMarks(content: str) -> str:
 | 
			
		|||
    """Replaces quotes with html formatting
 | 
			
		||||
    "hello" becomes <q>hello</q>
 | 
			
		||||
    """
 | 
			
		||||
    if '--BEGIN PGP MESSAGE--' in content or \
 | 
			
		||||
       '--BEGIN PGP PUBLIC KEY BLOCK--' in content:
 | 
			
		||||
        return content
 | 
			
		||||
    if '"' not in content:
 | 
			
		||||
        if '"' not in content:
 | 
			
		||||
            return content
 | 
			
		||||
| 
						 | 
				
			
			@ -197,6 +203,9 @@ def dangerousCSS(filename: str, allowLocalNetworkAccess: bool) -> bool:
 | 
			
		|||
def switchWords(baseDir: str, nickname: str, domain: str, content: str) -> str:
 | 
			
		||||
    """Performs word replacements. eg. Trump -> The Orange Menace
 | 
			
		||||
    """
 | 
			
		||||
    if '--BEGIN PGP MESSAGE--' in content or \
 | 
			
		||||
       '--BEGIN PGP PUBLIC KEY BLOCK--' in content:
 | 
			
		||||
        return content
 | 
			
		||||
    switchWordsFilename = baseDir + '/accounts/' + \
 | 
			
		||||
        nickname + '@' + domain + '/replacewords.txt'
 | 
			
		||||
    if not os.path.isfile(switchWordsFilename):
 | 
			
		||||
| 
						 | 
				
			
			@ -582,6 +591,9 @@ def _addMention(wordStr: str, httpPrefix: str, following: str, petnames: str,
 | 
			
		|||
def replaceContentDuplicates(content: str) -> str:
 | 
			
		||||
    """Replaces invalid duplicates within content
 | 
			
		||||
    """
 | 
			
		||||
    if '--BEGIN PGP MESSAGE--' in content or \
 | 
			
		||||
       '--BEGIN PGP PUBLIC KEY BLOCK--' in content:
 | 
			
		||||
        return content
 | 
			
		||||
    while '<<' in content:
 | 
			
		||||
        content = content.replace('<<', '<')
 | 
			
		||||
    while '>>' in content:
 | 
			
		||||
| 
						 | 
				
			
			@ -593,6 +605,9 @@ def replaceContentDuplicates(content: str) -> str:
 | 
			
		|||
def removeTextFormatting(content: str) -> str:
 | 
			
		||||
    """Removes markup for bold, italics, etc
 | 
			
		||||
    """
 | 
			
		||||
    if '--BEGIN PGP MESSAGE--' in content or \
 | 
			
		||||
       '--BEGIN PGP PUBLIC KEY BLOCK--' in content:
 | 
			
		||||
        return content
 | 
			
		||||
    if '<' not in content:
 | 
			
		||||
        return content
 | 
			
		||||
    removeMarkup = ('b', 'i', 'ul', 'ol', 'li', 'em', 'strong',
 | 
			
		||||
| 
						 | 
				
			
			@ -610,6 +625,9 @@ def removeLongWords(content: str, maxWordLength: int,
 | 
			
		|||
    """Breaks up long words so that on mobile screens this doesn't
 | 
			
		||||
    disrupt the layout
 | 
			
		||||
    """
 | 
			
		||||
    if '--BEGIN PGP MESSAGE--' in content or \
 | 
			
		||||
       '--BEGIN PGP PUBLIC KEY BLOCK--' in content:
 | 
			
		||||
        return content
 | 
			
		||||
    content = replaceContentDuplicates(content)
 | 
			
		||||
    if ' ' not in content:
 | 
			
		||||
        # handle a single very long string with no spaces
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -25,6 +25,7 @@ from follow import sendFollowRequestViaServer
 | 
			
		|||
from follow import sendUnfollowRequestViaServer
 | 
			
		||||
from posts import sendPostViaServer
 | 
			
		||||
from announce import sendAnnounceViaServer
 | 
			
		||||
from pgp import pgpDecrypt
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def _waitForKeypress(timeout: int, debug: bool) -> str:
 | 
			
		||||
| 
						 | 
				
			
			@ -485,10 +486,11 @@ def runNotificationsClient(baseDir: str, proxyType: str, httpPrefix: str,
 | 
			
		|||
                        else:
 | 
			
		||||
                            messageStr = speakerJson['say'] + '. ' + \
 | 
			
		||||
                                speakerJson['imageDescription']
 | 
			
		||||
                        messageStr = pgpDecrypt(messageStr)
 | 
			
		||||
 | 
			
		||||
                        content = messageStr
 | 
			
		||||
                        if speakerJson.get('content'):
 | 
			
		||||
                            content = speakerJson['content']
 | 
			
		||||
                            content = pgpDecrypt(speakerJson['content'])
 | 
			
		||||
 | 
			
		||||
                        # say the speaker's name
 | 
			
		||||
                        _sayCommand(nameStr, nameStr, screenreader,
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
							
								
								
									
										94
									
								
								pgp.py
								
								
								
								
							
							
						
						
									
										94
									
								
								pgp.py
								
								
								
								
							| 
						 | 
				
			
			@ -6,6 +6,8 @@ __maintainer__ = "Bob Mottram"
 | 
			
		|||
__email__ = "bob@freedombone.net"
 | 
			
		||||
__status__ = "Production"
 | 
			
		||||
 | 
			
		||||
import subprocess
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def getEmailAddress(actorJson: {}) -> str:
 | 
			
		||||
    """Returns the email address for the given actor
 | 
			
		||||
| 
						 | 
				
			
			@ -232,9 +234,9 @@ def extractPGPPublicKey(content: str) -> str:
 | 
			
		|||
    """
 | 
			
		||||
    startBlock = '--BEGIN PGP PUBLIC KEY BLOCK--'
 | 
			
		||||
    endBlock = '--END PGP PUBLIC KEY BLOCK--'
 | 
			
		||||
    if not startBlock in content:
 | 
			
		||||
    if startBlock not in content:
 | 
			
		||||
        return None
 | 
			
		||||
    if not endBlock in content:
 | 
			
		||||
    if endBlock not in content:
 | 
			
		||||
        return None
 | 
			
		||||
    if '\n' not in content:
 | 
			
		||||
        return None
 | 
			
		||||
| 
						 | 
				
			
			@ -252,3 +254,91 @@ def extractPGPPublicKey(content: str) -> str:
 | 
			
		|||
        if extracting:
 | 
			
		||||
            publicKey += line + '\n'
 | 
			
		||||
    return publicKey
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def _pgpImportPubKey(recipientPubKey: str) -> str:
 | 
			
		||||
    """ 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
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def pgpEncrypt(content: str, recipientPubKey: str) -> str:
 | 
			
		||||
    """ Encrypt using your default pgp key to the given recipient
 | 
			
		||||
    """
 | 
			
		||||
    keyId = _pgpImportPubKey(recipientPubKey)
 | 
			
		||||
    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')
 | 
			
		||||
    if '--BEGIN PGP MESSAGE--' not in encryptResult:
 | 
			
		||||
        return None
 | 
			
		||||
    return encryptResult
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def pgpDecrypt(content: str) -> str:
 | 
			
		||||
    """ Encrypt using your default pgp key to the given recipient
 | 
			
		||||
    """
 | 
			
		||||
    if '--BEGIN PGP MESSAGE--' not in content:
 | 
			
		||||
        return content
 | 
			
		||||
 | 
			
		||||
    # if the public key is also included within the message then import it
 | 
			
		||||
    startBlock = '--BEGIN PGP PUBLIC KEY BLOCK--'
 | 
			
		||||
    if startBlock in content:
 | 
			
		||||
        pubKey = extractPGPPublicKey(content)
 | 
			
		||||
        if pubKey:
 | 
			
		||||
            _pgpImportPubKey(pubKey)
 | 
			
		||||
 | 
			
		||||
    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
 | 
			
		||||
    decryptResult = decryptResult.decode('utf-8')
 | 
			
		||||
    return decryptResult
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
							
								
								
									
										33
									
								
								speaker.py
								
								
								
								
							
							
						
						
									
										33
									
								
								speaker.py
								
								
								
								
							| 
						 | 
				
			
			@ -413,21 +413,24 @@ def _postToSpeakerJson(baseDir: str, httpPrefix: str,
 | 
			
		|||
    content = urllib.parse.unquote_plus(postJsonObject['object']['content'])
 | 
			
		||||
    content = html.unescape(content)
 | 
			
		||||
    content = content.replace('<p>', '').replace('</p>', ' ')
 | 
			
		||||
    # replace some emoji before removing html
 | 
			
		||||
    if ' <3' in content:
 | 
			
		||||
        content = content.replace(' <3', ' ' + translate['heart'])
 | 
			
		||||
    content = removeHtml(htmlReplaceQuoteMarks(content))
 | 
			
		||||
    content = speakerReplaceLinks(content, translate, detectedLinks)
 | 
			
		||||
    # replace all double spaces
 | 
			
		||||
    while '  ' in content:
 | 
			
		||||
        content = content.replace('  ', ' ')
 | 
			
		||||
    content = content.replace(' . ', '. ').strip()
 | 
			
		||||
    sayContent = content
 | 
			
		||||
    sayContent = _speakerPronounce(baseDir, content, translate)
 | 
			
		||||
    # replace all double spaces
 | 
			
		||||
    while '  ' in sayContent:
 | 
			
		||||
        sayContent = sayContent.replace('  ', ' ')
 | 
			
		||||
    sayContent = sayContent.replace(' . ', '. ').strip()
 | 
			
		||||
    if '--BEGIN PGP MESSAGE--' not in content:
 | 
			
		||||
        # replace some emoji before removing html
 | 
			
		||||
        if ' <3' in content:
 | 
			
		||||
            content = content.replace(' <3', ' ' + translate['heart'])
 | 
			
		||||
        content = removeHtml(htmlReplaceQuoteMarks(content))
 | 
			
		||||
        content = speakerReplaceLinks(content, translate, detectedLinks)
 | 
			
		||||
        # replace all double spaces
 | 
			
		||||
        while '  ' in content:
 | 
			
		||||
            content = content.replace('  ', ' ')
 | 
			
		||||
        content = content.replace(' . ', '. ').strip()
 | 
			
		||||
        sayContent = content
 | 
			
		||||
        sayContent = _speakerPronounce(baseDir, content, translate)
 | 
			
		||||
        # replace all double spaces
 | 
			
		||||
        while '  ' in sayContent:
 | 
			
		||||
            sayContent = sayContent.replace('  ', ' ')
 | 
			
		||||
        sayContent = sayContent.replace(' . ', '. ').strip()
 | 
			
		||||
    else:
 | 
			
		||||
        sayContent = content
 | 
			
		||||
 | 
			
		||||
    imageDescription = ''
 | 
			
		||||
    if postJsonObject['object'].get('attachment'):
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
							
								
								
									
										5
									
								
								tests.py
								
								
								
								
							
							
						
						
									
										5
									
								
								tests.py
								
								
								
								
							| 
						 | 
				
			
			@ -3000,7 +3000,8 @@ def testFunctions():
 | 
			
		|||
        'E2EEremoveDevice',
 | 
			
		||||
        'setOrganizationScheme',
 | 
			
		||||
        'fill_headers',
 | 
			
		||||
        '_nothing'
 | 
			
		||||
        '_nothing',
 | 
			
		||||
        "pgpEncrypt"
 | 
			
		||||
    ]
 | 
			
		||||
    excludeImports = [
 | 
			
		||||
        'link',
 | 
			
		||||
| 
						 | 
				
			
			@ -3418,7 +3419,7 @@ def testEmojiImages():
 | 
			
		|||
def testExtractPGPPublicKey():
 | 
			
		||||
    print('testExtractPGPPublicKey')
 | 
			
		||||
    pubKey = \
 | 
			
		||||
        '-----BEGIN PGP PUBLIC KEY BLOCK-----\n' + \
 | 
			
		||||
        '-----BEGIN PGP PUBLIC KEY BLOCK-----\n\n' + \
 | 
			
		||||
        'mDMEWZBueBYJKwYBBAHaRw8BAQdAKx1t6wL0RTuU6/' + \
 | 
			
		||||
        'IBjngMbVJJ3Wg/3UW73/PV\n' + \
 | 
			
		||||
        'I47xKTS0IUJvYiBNb3R0cmFtIDxib2JAZnJlZWRvb' + \
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
		Loading…
	
		Reference in New Issue