nbuechner/synapse-invite-checker
nbuechner/synapse-invite-checker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

small fixes

Browse source
This commit is contained in:
Nils Büchner 2024-09-14 20:17:43 +02:00
parent 7912551d86
commit 45bdf91ef2
2 changed files with 70 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
README.md
View file

@ -1,11 +1,11 @@
# Synapse Invite Checker Module # Synapse Invite Checker Module
This is a Synapse module that checks incoming invites based on whitelist and blacklist rules. The module allows or blocks invites from certain homeservers depending on whether they appear in a dynamically fetched whitelist or blacklist JSON file. This is a Synapse module that checks incoming invites based on allowlist and blocklist rules. The module allows or blocks invites from certain homeservers depending on whether they appear in a dynamically fetched allowlist or blocklist JSON file.
## Features ## Features
- **Whitelist and Blacklist**: Allows invites from homeservers in the whitelist, blocks invites from homeservers in the blacklist. - **allowlist and blocklist**: Allows invites from homeservers in the allowlist, blocks invites from homeservers in the blocklist.
- **Dynamic Fetching**: The whitelist and blacklist are fetched dynamically from a provided URL, and cached for 10 minutes. - **Dynamic Fetching**: The allowlist and blocklist are fetched dynamically from a provided URL, and cached.
- **Fallback on Failure**: If the JSON file cannot be fetched (e.g., network error), the module automatically allows all invites to prevent disruptions. - **Fallback on Failure**: If the JSON file cannot be fetched (e.g., network error), the module automatically allows all invites to prevent disruptions.
## Configuration ## Configuration
@ -16,11 +16,9 @@ Add this module to your Synapse's `homeserver.yaml` under the `modules` section.
modules: modules:
- module: synapse_invite_checker.InviteChecker - module: synapse_invite_checker.InviteChecker
config: config:
# URL to fetch the JSON file containing the whitelist and blacklist # URL to fetch the JSON file containing the allowlist and blocklist
blacklist_whitelist_url: "https://example.com/invite-checker-lists.json" blocklist_allowlist_url: "https://example.com/invite-checker-lists.json"
# Whether to use the allowlist to allow certain homeservers (default: true)
# Whether to use the whitelist to allow certain homeservers (default: true) use_allowlist: true
use_whitelist: true # Whether to use the blocklist to block certain homeservers (default: true)
use_blocklist: true
# Whether to use the blacklist to block certain homeservers (default: true)
use_blacklist: true

116
synapse_invite_checker/invite_checker.py
View file

@ -1,4 +1,5 @@
import treq import treq
import json
from twisted.internet.defer import inlineCallbacks, returnValue from twisted.internet.defer import inlineCallbacks, returnValue
from cachetools import TTLCache from cachetools import TTLCache
from synapse.module_api import ModuleApi, errors from synapse.module_api import ModuleApi, errors
@ -9,10 +10,10 @@ logger = logging.getLogger(__name__)
class InviteCheckerConfig: class InviteCheckerConfig:
def __init__(self, config): def __init__(self, config):
# Initialize whitelist and blacklist toggle settings # Initialize allowlist and blocklist toggle settings
self.use_whitelist = config.get("use_whitelist", True) self.use_allowlist = config.get("use_allowlist", True)
self.use_blacklist = config.get("use_blacklist", True) self.use_blocklist = config.get("use_blocklist", True)
self.blacklist_whitelist_url = config.get("blacklist_whitelist_url", None) self.blocklist_allowlist_url = config.get("blocklist_allowlist_url", None)
@staticmethod @staticmethod
def parse_config(config): def parse_config(config):
@ -23,15 +24,15 @@ class InviteChecker:
self.api = api self.api = api
self.config = InviteCheckerConfig.parse_config(config) self.config = InviteCheckerConfig.parse_config(config)
# Initialize toggles for enabling or disabling whitelist and blacklist # Initialize toggles for enabling or disabling allowlist and blocklist
self.use_whitelist = self.config.use_whitelist self.use_allowlist = self.config.use_allowlist
self.use_blacklist = self.config.use_blacklist self.use_blocklist = self.config.use_blocklist
# Cache for whitelist/blacklist (TTL = 10 minutes) # Cache for allowlist/blocklist (TTL = 60 seconds)
self.cache = TTLCache(maxsize=1, ttl=600) self.cache = TTLCache(maxsize=1, ttl=60)
self.blacklist = set() self.blocklist = set()
self.whitelist = set() self.allowlist = set()
self.allow_all_invites_on_error = False # Flag to allow all invites if the JSON fetch fails self.allow_all_invites_on_error = False # Flag to allow all invites if the JSON fetch fails
# Register spam checker callback # Register spam checker callback
@ -40,14 +41,20 @@ class InviteChecker:
@inlineCallbacks @inlineCallbacks
def fetch_json(self, url): def fetch_json(self, url):
"""Fetch JSON data from the URL asynchronously using Twisted treq.""" """Fetch JSON data from a file download link asynchronously."""
logger.debug(f"Fetching JSON data from: {url}") logger.info(f"Fetching JSON data from: {url}")
try: try:
response = yield treq.get(url) response = yield treq.get(url)
if response.code == 200: if response.code == 200:
data = yield response.json() try:
logger.debug(f"Received data: {data}") # Try parsing the content as JSON, even if it's a file download
returnValue(data) content = yield response.content()
data = json.loads(content.decode('utf-8'))
logger.debug(f"Received data: {data}")
returnValue(data)
except Exception as json_error:
logger.error(f"Failed to decode JSON data: {json_error}")
returnValue(None)
else: else:
logger.error(f"Failed to fetch JSON data. Status code: {response.code}") logger.error(f"Failed to fetch JSON data. Status code: {response.code}")
returnValue(None) returnValue(None)
@ -55,73 +62,74 @@ class InviteChecker:
logger.error(f"Error while fetching JSON: {str(e)}") logger.error(f"Error while fetching JSON: {str(e)}")
returnValue(None) returnValue(None)
@inlineCallbacks
def update_blacklist_whitelist(self):
"""Fetch and update the blacklist and whitelist from the URLs."""
logger.info("Updating blacklist and whitelist")
json_data = yield self.fetch_json(self.config.blacklist_whitelist_url) @inlineCallbacks
def update_blocklist_allowlist(self):
"""Fetch and update the blocklist and allowlist from the URLs."""
logger.info("Updating blocklist and allowlist")
json_data = yield self.fetch_json(self.config.blocklist_allowlist_url)
if json_data: if json_data:
self.allow_all_invites_on_error = False # Reset the error flag if we successfully fetch the JSON self.allow_all_invites_on_error = False # Reset the error flag if we successfully fetch the JSON
self.use_whitelist = json_data.get('use_whitelist', True) self.use_allowlist = json_data.get('use_allowlist', True)
self.use_blacklist = json_data.get('use_blacklist', True) self.use_blocklist = json_data.get('use_blocklist', True)
self.blacklist = set(json_data.get('blacklist', [])) self.blocklist = set(json_data.get('blocklist', []))
self.whitelist = set(json_data.get('whitelist', [])) self.allowlist = set(json_data.get('allowlist', []))
# Cache the data # Cache the data
self.cache['blacklist_whitelist'] = (self.blacklist, self.whitelist) self.cache['blocklist_allowlist'] = (self.blocklist, self.allowlist)
logger.info(f"Updated whitelist: {self.whitelist}") logger.info(f"Updated allowlist: {self.allowlist}")
logger.info(f"Updated blacklist: {self.blacklist}") logger.info(f"Updated blocklist: {self.blocklist}")
else: else:
# Set the error flag to allow all invites if fetching the JSON fails # Set the error flag to allow all invites if fetching the JSON fails
logger.error("Failed to update whitelist/blacklist due to missing JSON data. Allowing all invites.") logger.error("Failed to update allowlist/blocklist due to missing JSON data. Allowing all invites.")
self.allow_all_invites_on_error = True self.allow_all_invites_on_error = True
@inlineCallbacks @inlineCallbacks
def get_blacklist_whitelist(self): def get_blocklist_allowlist(self):
"""Return cached blacklist/whitelist or fetch new data if cache is expired.""" """Return cached blocklist/allowlist or fetch new data if cache is expired."""
if self.allow_all_invites_on_error: if self.allow_all_invites_on_error:
logger.info("Skipping whitelist/blacklist checks because of previous JSON fetch failure.") logger.info("Skipping allowlist/blocklist checks because of previous JSON fetch failure.")
returnValue((set(), set())) # Return empty sets to indicate no checks should be applied returnValue((set(), set())) # Return empty sets to indicate no checks should be applied
try: try:
returnValue(self.cache['blacklist_whitelist']) returnValue(self.cache['blocklist_allowlist'])
except KeyError: except KeyError:
yield self.update_blacklist_whitelist() yield self.update_blocklist_allowlist()
returnValue(self.cache['blacklist_whitelist']) returnValue(self.cache['blocklist_allowlist'])
@inlineCallbacks @inlineCallbacks
def user_may_invite(self, inviter: str, invitee: str, room_id: str): def user_may_invite(self, inviter: str, invitee: str, room_id: str):
"""Check if a user may invite another based on whitelist/blacklist rules.""" """Check if a user may invite another based on allowlist/blocklist rules."""
logger.debug(f"Checking invite from {inviter} to {invitee}") logger.info(f"Checking invite from {inviter} to {invitee}")
# If JSON fetching failed, allow all invites # If JSON fetching failed, allow all invites
if self.allow_all_invites_on_error: if self.allow_all_invites_on_error:
logger.info(f"Allowing invite from {inviter} to {invitee} due to previous JSON fetch failure.") logger.info(f"Allowing invite from {inviter} to {invitee} due to previous JSON fetch failure.")
returnValue("NOT_SPAM") # Allow all invites returnValue("NOT_SPAM") # Allow all invites
# Proceed with whitelist/blacklist checks if JSON fetching was successful # Proceed with allowlist/blocklist checks if JSON fetching was successful
blacklist, whitelist = yield self.get_blacklist_whitelist() blocklist, allowlist = yield self.get_blocklist_allowlist()
inviter_domain = UserID.from_string(inviter).domain inviter_domain = UserID.from_string(inviter).domain
# Whitelist check (if enabled) # allowlist check (if enabled)
if self.use_whitelist: if self.use_allowlist:
logger.debug(f"Whitelist enabled. Checking domain: {inviter_domain}") logger.info(f"allowlist enabled. Checking domain: {inviter_domain}")
if inviter_domain in whitelist: if inviter_domain in allowlist:
logger.info(f"Invite allowed: {inviter_domain} is in the whitelist") logger.info(f"Invite allowed: {inviter_domain} is in the allowlist")
returnValue("NOT_SPAM") # Allow if the domain is in the whitelist returnValue("NOT_SPAM") # Allow if the domain is in the allowlist
# Blacklist check (if enabled) # blocklist check (if enabled)
if self.use_blacklist: if self.use_blocklist:
logger.debug(f"Blacklist enabled. Checking domain: {inviter_domain}") logger.info(f"blocklist enabled. Checking domain: {inviter_domain}")
if inviter_domain in blacklist: if inviter_domain in blocklist:
logger.info(f"Invite blocked: {inviter_domain} is in the blacklist") logger.info(f"Invite blocked: {inviter_domain} is in the blocklist")
returnValue(errors.Codes.FORBIDDEN) # Deny if the domain is in the blacklist returnValue(errors.Codes.FORBIDDEN) # Deny if the domain is in the blocklist
# If whitelist is enabled and domain is not in the whitelist, deny the invite # If allowlist is enabled and domain is not in the allowlist, deny the invite
if self.use_whitelist and inviter_domain not in whitelist: if self.use_allowlist and inviter_domain not in allowlist:
logger.info(f"Invite blocked: {inviter_domain} is not in the whitelist") logger.info(f"Invite blocked: {inviter_domain} is not in the allowlist")
returnValue(errors.Codes.FORBIDDEN) returnValue(errors.Codes.FORBIDDEN)
# Allow by default # Allow by default