epicyon/blocking.py

327 lines
12 KiB
Python
Raw Normal View History

2020-04-01 20:06:27 +00:00
__filename__ = "blocking.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "1.1.0"
__maintainer__ = "Bob Mottram"
__email__ = "bob@freedombone.net"
__status__ = "Production"
2019-07-14 19:27:13 +00:00
import os
2019-09-09 15:53:23 +00:00
from utils import isEvil
2020-04-01 20:06:27 +00:00
from utils import locatePost
2020-03-28 10:33:04 +00:00
from utils import evilIncarnate
2020-04-01 20:06:27 +00:00
from utils import getDomainFromActor
from utils import getNicknameFromActor
2019-07-14 19:27:13 +00:00
2020-04-01 20:06:27 +00:00
def addGlobalBlock(baseDir: str,
blockNickname: str, blockDomain: str) -> bool:
"""Global block which applies to all accounts
"""
2020-04-01 20:06:27 +00:00
blockingFilename = baseDir + '/accounts/blocking.txt'
2020-03-22 21:16:02 +00:00
if not blockNickname.startswith('#'):
2020-04-01 20:06:27 +00:00
blockHandle = blockNickname + '@' + blockDomain
2019-08-14 10:32:15 +00:00
if os.path.isfile(blockingFilename):
if blockHandle in open(blockingFilename).read():
return False
2020-04-01 20:06:27 +00:00
blockFile = open(blockingFilename, "a+")
blockFile.write(blockHandle + '\n')
2019-08-14 10:32:15 +00:00
blockFile.close()
else:
2020-04-01 20:06:27 +00:00
blockHashtag = blockNickname
2019-08-14 10:32:15 +00:00
if os.path.isfile(blockingFilename):
2020-04-01 20:06:27 +00:00
if blockHashtag + '\n' in open(blockingFilename).read():
2019-08-14 10:32:15 +00:00
return False
2020-04-01 20:06:27 +00:00
blockFile = open(blockingFilename, "a+")
blockFile.write(blockHashtag + '\n')
2019-08-14 10:32:15 +00:00
blockFile.close()
return True
2020-04-01 20:06:27 +00:00
def addBlock(baseDir: str, nickname: str, domain: str,
blockNickname: str, blockDomain: str) -> bool:
2019-07-14 19:27:13 +00:00
"""Block the given account
"""
2019-07-17 21:40:56 +00:00
if ':' in domain:
2020-04-01 20:06:27 +00:00
domain = domain.split(':')[0]
blockingFilename = baseDir + '/accounts/' + \
nickname + '@' + domain + '/blocking.txt'
blockHandle = blockNickname + '@' + blockDomain
2019-07-14 19:27:13 +00:00
if os.path.isfile(blockingFilename):
if blockHandle in open(blockingFilename).read():
2019-07-14 19:57:05 +00:00
return False
2020-04-01 20:06:27 +00:00
blockFile = open(blockingFilename, "a+")
blockFile.write(blockHandle + '\n')
2019-07-14 19:27:13 +00:00
blockFile.close()
2019-07-14 19:57:05 +00:00
return True
2019-07-14 19:27:13 +00:00
2020-04-01 20:06:27 +00:00
def removeGlobalBlock(baseDir: str,
unblockNickname: str,
unblockDomain: str) -> bool:
"""Unblock the given global block
"""
2020-04-01 20:06:27 +00:00
unblockingFilename = baseDir + '/accounts/blocking.txt'
2020-03-22 21:16:02 +00:00
if not unblockNickname.startswith('#'):
2020-04-01 20:06:27 +00:00
unblockHandle = unblockNickname + '@' + unblockDomain
2019-08-14 10:32:15 +00:00
if os.path.isfile(unblockingFilename):
if unblockHandle in open(unblockingFilename).read():
with open(unblockingFilename, 'r') as fp:
2020-04-01 20:06:27 +00:00
with open(unblockingFilename + '.new', 'w') as fpnew:
2019-08-14 10:32:15 +00:00
for line in fp:
2020-04-01 20:06:27 +00:00
handle = line.replace('\n', '')
2019-08-14 10:32:15 +00:00
if unblockHandle not in line:
2020-04-01 20:06:27 +00:00
fpnew.write(handle + '\n')
if os.path.isfile(unblockingFilename + '.new'):
os.rename(unblockingFilename + '.new', unblockingFilename)
2019-08-14 10:32:15 +00:00
return True
else:
2020-04-01 20:06:27 +00:00
unblockHashtag = unblockNickname
2019-08-14 10:32:15 +00:00
if os.path.isfile(unblockingFilename):
2020-04-01 20:06:27 +00:00
if unblockHashtag + '\n' in open(unblockingFilename).read():
2019-08-14 10:32:15 +00:00
with open(unblockingFilename, 'r') as fp:
2020-04-01 20:06:27 +00:00
with open(unblockingFilename + '.new', 'w') as fpnew:
2019-08-14 10:32:15 +00:00
for line in fp:
2020-04-01 20:06:27 +00:00
blockLine = line.replace('\n', '')
2019-08-14 10:32:15 +00:00
if unblockHashtag not in line:
2020-04-01 20:06:27 +00:00
fpnew.write(blockLine + '\n')
if os.path.isfile(unblockingFilename + '.new'):
os.rename(unblockingFilename + '.new', unblockingFilename)
2019-08-14 10:32:15 +00:00
return True
return False
2020-04-01 20:06:27 +00:00
def removeBlock(baseDir: str, nickname: str, domain: str,
unblockNickname: str, unblockDomain: str) -> bool:
2019-07-14 19:27:13 +00:00
"""Unblock the given account
"""
2019-07-17 21:40:56 +00:00
if ':' in domain:
2020-04-01 20:06:27 +00:00
domain = domain.split(':')[0]
unblockingFilename = baseDir + '/accounts/' + \
nickname + '@' + domain + '/blocking.txt'
unblockHandle = unblockNickname + '@' + unblockDomain
2019-07-14 19:27:13 +00:00
if os.path.isfile(unblockingFilename):
if unblockHandle in open(unblockingFilename).read():
with open(unblockingFilename, 'r') as fp:
2020-04-01 20:06:27 +00:00
with open(unblockingFilename + '.new', 'w') as fpnew:
2019-07-14 19:27:13 +00:00
for line in fp:
2020-04-01 20:06:27 +00:00
handle = line.replace('\n', '')
2019-07-14 19:27:13 +00:00
if unblockHandle not in line:
2020-04-01 20:06:27 +00:00
fpnew.write(handle + '\n')
if os.path.isfile(unblockingFilename + '.new'):
os.rename(unblockingFilename + '.new', unblockingFilename)
2019-07-14 19:57:05 +00:00
return True
return False
2019-08-14 10:32:15 +00:00
2020-04-01 20:06:27 +00:00
def isBlockedHashtag(baseDir: str, hashtag: str) -> bool:
2019-08-14 10:32:15 +00:00
"""Is the given hashtag blocked?
"""
2020-04-01 20:06:27 +00:00
globalBlockingFilename = baseDir + '/accounts/blocking.txt'
2019-08-14 10:32:15 +00:00
if os.path.isfile(globalBlockingFilename):
2020-04-01 20:06:27 +00:00
hashtag = hashtag.strip('\n')
if hashtag + '\n' in open(globalBlockingFilename).read():
2019-08-14 10:32:15 +00:00
return True
return False
2020-04-01 20:06:27 +00:00
2020-03-28 10:33:04 +00:00
def getDomainBlocklist(baseDir: str) -> str:
"""Returns all globally blocked domains as a string
This can be used for fast matching to mitigate flooding
"""
2020-04-01 20:06:27 +00:00
blockedStr = ''
2020-03-28 10:33:04 +00:00
2020-04-01 20:06:27 +00:00
evilDomains = evilIncarnate()
2020-03-28 10:33:04 +00:00
for evil in evilDomains:
2020-04-01 20:06:27 +00:00
blockedStr += evil + '\n'
2020-03-28 10:33:04 +00:00
2020-04-01 20:06:27 +00:00
globalBlockingFilename = baseDir + '/accounts/blocking.txt'
2020-03-28 10:33:04 +00:00
if not os.path.isfile(globalBlockingFilename):
return blockedStr
with open(globalBlockingFilename, 'r') as file:
blockedStr += file.read()
return blockedStr
2020-04-01 20:06:27 +00:00
def isBlockedDomain(baseDir: str, domain: str) -> bool:
"""Is the given domain blocked?
"""
if isEvil(domain):
return True
2020-04-01 20:06:27 +00:00
globalBlockingFilename = baseDir + '/accounts/blocking.txt'
if os.path.isfile(globalBlockingFilename):
2020-04-01 20:06:27 +00:00
if '*@' + domain in open(globalBlockingFilename).read():
return True
return False
2020-04-01 20:06:27 +00:00
def isBlocked(baseDir: str, nickname: str, domain: str,
blockNickname: str, blockDomain: str) -> bool:
2019-07-14 19:27:13 +00:00
"""Is the given nickname blocked?
"""
2019-09-09 15:53:23 +00:00
if isEvil(blockDomain):
return True
2020-04-01 20:06:27 +00:00
globalBlockingFilename = baseDir + '/accounts/blocking.txt'
if os.path.isfile(globalBlockingFilename):
2020-04-01 20:06:27 +00:00
if '*@' + blockDomain in open(globalBlockingFilename).read():
return True
2020-02-05 17:41:24 +00:00
if blockNickname:
2020-04-01 20:06:27 +00:00
blockHandle = blockNickname + '@' + blockDomain
2020-02-05 17:39:41 +00:00
if blockHandle in open(globalBlockingFilename).read():
return True
2020-04-01 20:06:27 +00:00
allowFilename = baseDir + '/accounts/' + \
nickname + '@' + domain + '/allowedinstances.txt'
if os.path.isfile(allowFilename):
if blockDomain not in open(allowFilename).read():
return True
2020-04-01 20:06:27 +00:00
blockingFilename = baseDir + '/accounts/' + \
nickname + '@' + domain + '/blocking.txt'
2019-07-14 19:27:13 +00:00
if os.path.isfile(blockingFilename):
2020-04-01 20:06:27 +00:00
if '*@' + blockDomain in open(blockingFilename).read():
2019-08-02 11:46:42 +00:00
return True
2020-02-05 17:41:24 +00:00
if blockNickname:
2020-04-01 20:06:27 +00:00
blockHandle = blockNickname + '@' + blockDomain
2020-02-05 17:39:41 +00:00
if blockHandle in open(blockingFilename).read():
return True
2019-07-14 19:27:13 +00:00
return False
2020-04-01 20:06:27 +00:00
def outboxBlock(baseDir: str, httpPrefix: str,
nickname: str, domain: str, port: int,
messageJson: {}, debug: bool) -> None:
2019-07-17 21:40:56 +00:00
""" When a block request is received by the outbox from c2s
"""
if not messageJson.get('type'):
if debug:
print('DEBUG: block - no type')
return
2020-04-01 20:06:27 +00:00
if not messageJson['type'] == 'Block':
2019-07-17 21:40:56 +00:00
if debug:
print('DEBUG: not a block')
return
if not messageJson.get('object'):
if debug:
print('DEBUG: no object in block')
return
if not isinstance(messageJson['object'], str):
if debug:
print('DEBUG: block object is not string')
return
if debug:
print('DEBUG: c2s block request arrived in outbox')
2020-04-01 20:06:27 +00:00
messageId = messageJson['object'].replace('/activity', '')
2019-07-17 21:40:56 +00:00
if '/statuses/' not in messageId:
if debug:
print('DEBUG: c2s block object is not a status')
return
2019-10-17 22:26:47 +00:00
if '/users/' not in messageId and \
'/channel/' not in messageId and \
'/profile/' not in messageId:
2019-07-17 21:40:56 +00:00
if debug:
print('DEBUG: c2s block object has no nickname')
return
if ':' in domain:
2020-04-01 20:06:27 +00:00
domain = domain.split(':')[0]
postFilename = locatePost(baseDir, nickname, domain, messageId)
2019-07-17 21:40:56 +00:00
if not postFilename:
if debug:
print('DEBUG: c2s block post not found in inbox or outbox')
print(messageId)
2019-09-02 09:43:43 +00:00
return
2020-04-01 20:06:27 +00:00
nicknameBlocked = getNicknameFromActor(messageJson['object'])
2019-09-02 09:43:43 +00:00
if not nicknameBlocked:
2020-04-01 20:06:27 +00:00
print('WARN: unable to find nickname in ' + messageJson['object'])
2019-09-02 09:43:43 +00:00
return
2020-04-01 20:06:27 +00:00
domainBlocked, portBlocked = getDomainFromActor(messageJson['object'])
domainBlockedFull = domainBlocked
2019-07-17 21:40:56 +00:00
if portBlocked:
2020-04-01 20:06:27 +00:00
if portBlocked != 80 and portBlocked != 443:
if ':' not in domainBlocked:
2020-04-01 20:06:27 +00:00
domainBlockedFull = domainBlocked + ':' + str(portBlocked)
2019-07-17 21:40:56 +00:00
2020-04-01 20:06:27 +00:00
addBlock(baseDir, nickname, domain,
nicknameBlocked, domainBlockedFull)
2020-03-22 20:59:01 +00:00
2019-07-17 21:40:56 +00:00
if debug:
2020-04-01 20:06:27 +00:00
print('DEBUG: post blocked via c2s - ' + postFilename)
2019-07-17 21:40:56 +00:00
2020-04-01 20:06:27 +00:00
def outboxUndoBlock(baseDir: str, httpPrefix: str,
nickname: str, domain: str, port: int,
messageJson: {}, debug: bool) -> None:
2019-07-17 21:40:56 +00:00
""" When an undo block request is received by the outbox from c2s
"""
if not messageJson.get('type'):
if debug:
print('DEBUG: undo block - no type')
return
2020-04-01 20:06:27 +00:00
if not messageJson['type'] == 'Undo':
2019-07-17 21:40:56 +00:00
if debug:
print('DEBUG: not an undo block')
return
if not messageJson.get('object'):
if debug:
print('DEBUG: no object in undo block')
return
if not isinstance(messageJson['object'], dict):
if debug:
print('DEBUG: undo block object is not string')
return
if not messageJson['object'].get('type'):
if debug:
print('DEBUG: undo block - no type')
return
2020-04-01 20:06:27 +00:00
if not messageJson['object']['type'] == 'Block':
2019-07-17 21:40:56 +00:00
if debug:
print('DEBUG: not an undo block')
return
if not messageJson['object'].get('object'):
if debug:
print('DEBUG: no object in undo block')
return
if not isinstance(messageJson['object']['object'], str):
if debug:
print('DEBUG: undo block object is not string')
return
if debug:
print('DEBUG: c2s undo block request arrived in outbox')
2020-04-01 20:06:27 +00:00
messageId = messageJson['object']['object'].replace('/activity', '')
2019-07-17 21:40:56 +00:00
if '/statuses/' not in messageId:
if debug:
print('DEBUG: c2s undo block object is not a status')
return
2019-10-17 22:26:47 +00:00
if '/users/' not in messageId and \
'/channel/' not in messageId and \
'/profile/' not in messageId:
2019-07-17 21:40:56 +00:00
if debug:
print('DEBUG: c2s undo block object has no nickname')
return
if ':' in domain:
2020-04-01 20:06:27 +00:00
domain = domain.split(':')[0]
postFilename = locatePost(baseDir, nickname, domain, messageId)
2019-07-17 21:40:56 +00:00
if not postFilename:
if debug:
print('DEBUG: c2s undo block post not found in inbox or outbox')
print(messageId)
2019-09-02 09:43:43 +00:00
return
2020-04-01 20:06:27 +00:00
nicknameBlocked = getNicknameFromActor(messageJson['object']['object'])
2019-09-02 09:43:43 +00:00
if not nicknameBlocked:
2020-04-01 20:06:27 +00:00
print('WARN: unable to find nickname in ' +
2020-03-30 19:09:45 +00:00
messageJson['object']['object'])
2019-09-02 09:43:43 +00:00
return
2020-04-01 20:06:27 +00:00
domainObject = messageJson['object']['object']
domainBlocked, portBlocked = getDomainFromActor(domainObject)
domainBlockedFull = domainBlocked
2019-07-17 21:40:56 +00:00
if portBlocked:
2020-04-01 20:06:27 +00:00
if portBlocked != 80 and portBlocked != 443:
if ':' not in domainBlocked:
2020-04-01 20:06:27 +00:00
domainBlockedFull = domainBlocked + ':' + str(portBlocked)
2019-07-17 21:40:56 +00:00
2020-04-01 20:06:27 +00:00
removeBlock(baseDir, nickname, domain,
nicknameBlocked, domainBlockedFull)
2019-07-17 21:40:56 +00:00
if debug:
2020-04-01 20:06:27 +00:00
print('DEBUG: post undo blocked via c2s - ' + postFilename)