thecoffemaker/Slixfeed
1
0
Fork 0
forked from sch/Slixfeed
Code Pull requests Activity

Add a selection mechanism to manage whitelist and blacklist.

Browse source
This commit is contained in:
Schimon Jehudah, Adv. 2024-11-15 13:34:26 +02:00
parent 709bac58b8
commit 09bd838ad2
5 changed files with 143 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
slixfeed/assets/selector.toml Normal file
View file

@ -0,0 +1,27 @@
enabled = 0
blacklist = [
"christian@iceagefarmer.com",
"david@falkon.org",
"diggy@diggy.club",
"dino@drdino.com",
"doug@baxterisland.com",
"doug@blacklistednews.com",
"emdek@otter-browser.org",
"eric@drdino.com",
"gemini@circumlunar.space",
"hal@reallibertymedia.com",
"henrik@redice.tv",
"jg@cwns.i2p",
"jo@drdino.com",
"joel@thelunaticfarmer.com",
"kent@drdino.com",
"lana@redice.tv",
"larken@larkenrose.com",
"lee@oraclebroadcasting.com",
"mark@marksargent.com",
"nick@nightnationreview.com",
"oliver@postmarketos.org",
"robert@cwns.i2p",
"patrick@slackware.com",
]
whitelist = []

5
slixfeed/config.py
View file

@ -51,6 +51,11 @@ logger = Logger(__name__)
# setting_jid.setting_key has value, otherwise resort to setting_default.setting_key.
class Config:
def update_toml_file(filename, data):
with open(filename, 'w') as new_file:
content = tomli_w.dumps(data)
new_file.write(content)
def add_settings_default(self):
settings_default = get_values('settings.toml', 'settings')
self.defaults = settings_default

40
slixfeed/xmpp/chat.py
View file

@ -596,6 +596,44 @@ class XmppChat:
case _ if command_lowercase.startswith('search'):
query = command[7:]
response = XmppCommands.search_items(db_file, query)
case _ if command_lowercase.startswith('blacklist'):
if XmppUtilities.is_operator(self, jid_bare):
action_jid = command[9:].strip()
action_jid_split = action_jid.split(' ')
if len(action_jid_split) == 2:
action, jid = action_jid_split
if jid and action == 'add':
response = XmppCommands.add_jid_to_selector(self, jid, 'blacklist')
elif jid and action == 'delete':
response = XmppCommands.del_jid_from_selector(self, jid, 'blacklist')
else:
response = f'Unknown action {action}.'
elif len(action_jid_split) > 2:
response = 'USAGE: blacklist <action> <jid>'
else:
response = XmppCommands.print_selector(self.blacklist)
else:
response = ('This action is restricted. '
'Type: managing blacklist.')
case _ if command_lowercase.startswith('whitelist'):
if XmppUtilities.is_operator(self, jid_bare):
action_jid = command[9:].strip()
action_jid_split = action_jid.split(' ')
if len(action_jid_split) == 2:
action, jid = action_jid_split
if jid and action == 'add':
response = XmppCommands.add_jid_to_selector(self, jid, 'whitelist')
elif jid and action == 'delete':
response = XmppCommands.del_jid_from_selector(self, jid, 'whitelist')
else:
response = f'Unknown action {action}.'
elif len(action_jid_split) > 2:
response = 'USAGE: whitelist <action> <jid>'
else:
response = XmppCommands.print_selector(self.whitelist)
else:
response = ('This action is restricted. '
'Type: managing blacklist.')
case 'start':
status_type = 'available'
status_message = '📫️ Welcome back.'
@ -891,7 +929,7 @@ class XmppChatAction:
# TODO Limit text length
# summary = summary.replace("\n\n\n", "\n\n")
summary = summary.replace('\n', ' ')
summary = summary.replace(' ', ' ')
summary = summary.replace(' ', ' ')
# summary = summary.replace(' ', ' ')
summary = ' '.join(summary.split())
length = Config.get_setting_value(self, jid, 'length')

37
slixfeed/xmpp/client.py
View file

@ -124,6 +124,12 @@ class XmppClient(slixmpp.ClientXMPP):
# Handler for operators
self.operators = config.get_values('accounts.toml', 'xmpp')['operators']
# Handlers for whitelist and blacklist
self.selector = config.get_values('selector.toml')
paywall_enabled = self.selector['enabled']
self.whitelist = self.selector['whitelist']
self.blacklist = self.selector['blacklist']
# self.settings = {}
# # Populate dict handler
# Config.add_settings_default(self)
@ -367,6 +373,9 @@ class XmppClient(slixmpp.ClientXMPP):
function_name = sys._getframe().f_code.co_name
message_log = '{}'
logger.debug(message_log.format(function_name))
# TODO Iterate self.blacklist and add JIDs to Privacy List
status_message = 'Slixfeed version {}'.format(__version__)
self.adhoc_commands()
for operator in self.operators:
@ -537,15 +546,25 @@ class XmppClient(slixmpp.ClientXMPP):
logger.debug(message_log.format(function_name, jid_full))
jid_bare = presence['from'].bare
if not self.client_roster[jid_bare]['to']:
# XmppPresence.subscription(self, jid, 'subscribe')
XmppPresence.subscription(self, jid_bare, 'subscribed')
await XmppRoster.add(self, jid_bare)
status_message = '✒️ Share online status to receive updates'
XmppPresence.send(self, jid_bare, status_message)
message_subject = 'RSS News Bot'
message_body = 'Share online status to receive updates.'
XmppMessage.send_headline(self, jid_bare, message_subject,
message_body, 'chat')
if not paywall_enabled or paywall_enabled and jid_bare in self.whitelist:
# XmppPresence.subscription(self, jid, 'subscribe')
XmppPresence.subscription(self, jid_bare, 'subscribed')
await XmppRoster.add(self, jid_bare)
status_message = '✒️ Share online status to receive updates'
XmppPresence.send(self, jid_bare, status_message)
message_subject = 'RSS News Bot'
message_body = 'Share online status to receive updates.'
XmppMessage.send_headline(self, jid_bare, message_subject,
message_body, 'chat')
else:
message_body = 'Subscription has been denied.\nContact an operator.'
XmppMessage.send_headline(self, jid_bare, message_subject,
message_body, 'chat')
message_body = f'Subscription has been denied for Jabber ID xmpp:{jid_bare}.'
for operator in self.operators:
XmppMessage.send_headline(self, operator['jid'],
message_subject, message_body,
'chat')
time_end = time.time()
difference = time_end - time_begin
if difference > 1: logger.warning('{} (time: {})'.format(function_name,

45
slixfeed/xmpp/commands.py
View file

@ -2,6 +2,7 @@
# -*- coding: utf-8 -*-
from feedparser import parse
import os
from random import randrange
import slixfeed.config as config
from slixfeed.config import Config
@ -18,6 +19,7 @@ from slixfeed.xmpp.presence import XmppPresence
from slixfeed.xmpp.status import XmppStatusTask
from slixfeed.xmpp.utilities import XmppUtilities
import sys
import tomli_w
try:
import tomllib
@ -1088,11 +1090,52 @@ class XmppCommands:
return message
def add_jid_to_selector(self, jid, list_type):
config_dir = config.get_default_config_directory()
filename = os.path.join(config_dir, 'selector.toml')
match list_type:
case 'blacklist':
list_type_list = self.blacklist
case 'whitelist':
list_type_list = self.whitelist
if jid in list_type_list:
message = f'Jabber ID {jid} is already included in {list_type}.\nNo action has been committed.'
else:
list_type_list.append(jid)
Config.update_toml_file(filename, self.selector)
message = f'Jabber ID {jid} has been added to {list_type}.'
return message
def del_jid_from_selector(self, jid, list_type):
config_dir = config.get_default_config_directory()
filename = os.path.join(config_dir, 'selector.toml')
match list_type:
case 'blacklist':
list_type_list = self.blacklist
case 'whitelist':
list_type_list = self.whitelist
if jid in list_type_list:
list_type_list.remove(jid)
Config.update_toml_file(filename, self.selector)
message = f'Jabber ID "{jid}" has been removed from {list_type}.'
else:
message = f'Jabber ID "{jid}" was not found in {list_type}.\nNo action has been committed.'
return message
def print_selector(list_type):
jids = ' '.join(list_type) if list_type else '(empty).'
message = f'Jabber IDs: {jids}'
return message
def print_support_jid():
muc_jid = 'slixfeed@chat.woodpeckersnest.space'
message = 'Join xmpp:{}?join'.format(muc_jid)
return message
async def invite_jid_to_muc(self, jid_bare):
muc_jid = 'slixfeed@chat.woodpeckersnest.space'
if await XmppUtilities.get_chat_type(self, jid_bare) == 'chat':