# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
+import logging
+
import time
-import validators
-from fba import blacklist
-from fba import fba
+from fba import database
+
+from fba.helpers import blacklist
+from fba.helpers import domain as domain_helper
from fba.helpers import tidyup
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
def update_reason(reason: str, blocker: str, blocked: str, block_level: str):
- # DEBUG: print(f"DEBUG: reason='{reason}',blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
+ logger.debug("reason='%s',blocker='%s',blocked='%s',block_level='%s' - CALLED!", reason, blocker, blocked, block_level)
+ domain_helper.raise_on(blocker)
+ domain_helper.raise_on(blocked)
+
if not isinstance(reason, str) and reason is not None:
- raise ValueError(f"Parameter reason[]='{type(reason)}' is not 'str'")
- elif not isinstance(blocker, str):
- raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
- elif blocker == "":
- raise ValueError("Parameter 'blocker' is empty")
- elif not isinstance(blocked, str):
- raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
- elif blocked == "":
- raise ValueError("Parameter 'blocked' is empty")
+ raise ValueError(f"Parameter reason[]='{type(reason)}' is not of type 'str'")
elif not isinstance(block_level, str):
- raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
+ raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
elif block_level == "":
raise ValueError("Parameter 'block_level' is empty")
- elif block_level == "accept":
- raise ValueError("Accepted domains are not wanted here")
+ elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
+ raise ValueError(f"block_level='{block_level}' is not wanted.")
+ elif blacklist.is_blacklisted(blocker):
+ raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
+ elif blacklist.is_blacklisted(blocked):
+ raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
+ elif not is_instance_blocked(blocker, blocked, block_level):
+ raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' is not registered but function is invoked")
- # DEBUG: print("DEBUG: Updating block reason:", reason, blocker, blocked, block_level)
- fba.cursor.execute(
+ logger.debug("Updating block reason='%s',blocker='%s',blocked='%s',block_level='%s'", reason, blocker, blocked, block_level)
+ database.cursor.execute(
"UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND (reason IS NULL OR reason = '') LIMIT 1",
[
reason,
block_level
])
- # DEBUG: print("DEBUG: EXIT!")
+ logger.debug("EXIT!")
def update_last_seen(blocker: str, blocked: str, block_level: str):
- # DEBUG: print("DEBUG: Updating last_seen for:", blocker, blocked, block_level)
- if not isinstance(blocker, str):
- raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
- elif blocker == "":
- raise ValueError("Parameter 'blocker' is empty")
- elif not isinstance(blocked, str):
- raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
- elif blocked == "":
- raise ValueError("Parameter 'blocked' is empty")
- elif not isinstance(block_level, str):
- raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
+ logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
+ domain_helper.raise_on(blocker)
+ domain_helper.raise_on(blocked)
+
+ if not isinstance(block_level, str):
+ raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
elif block_level == "":
raise ValueError("Parameter 'block_level' is empty")
- elif block_level == "accept":
- raise ValueError("Accepted domains are not wanted here")
+ elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
+ raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
+ elif blacklist.is_blacklisted(blocker):
+ raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
+ elif blacklist.is_blacklisted(blocked):
+ raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
+ elif not is_instance_blocked(blocker, blocked, block_level):
+ raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' is not registered but function is invoked")
- fba.cursor.execute(
+ database.cursor.execute(
"UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
[
time.time(),
block_level
])
- # DEBUG: print("DEBUG: EXIT!")
-
-def is_instance_blocked(blocker: str, blocked: str, block_level: str) -> bool:
- # DEBUG: print(f"DEBUG: blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
- if not isinstance(blocker, str):
- raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not of type 'str'")
- elif blocker == "":
- raise ValueError("Parameter 'blocker' is empty")
- elif not isinstance(blocked, str):
- raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not of type 'str'")
- elif blocked == "":
- raise ValueError("Parameter 'blocked' is empty")
- elif not isinstance(block_level, str):
+ logger.debug("EXIT!")
+
+def is_instance_blocked(blocker: str, blocked: str, block_level: str = None) -> bool:
+ logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
+ domain_helper.raise_on(blocker)
+ domain_helper.raise_on(blocked)
+
+ if not isinstance(block_level, str) and block_level is not None:
raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
elif block_level == "":
raise ValueError("Parameter 'block_level' is empty")
- elif block_level == "accept":
- raise ValueError("Accepted domains are not wanted here")
-
- fba.cursor.execute(
- "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
- (
- blocker,
- blocked,
- block_level
- ),
- )
-
- is_blocked = fba.cursor.fetchone() is not None
+ elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
+ raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
+ elif blacklist.is_blacklisted(blocker):
+ raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
+ elif blacklist.is_blacklisted(blocked):
+ raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
- # DEBUG: print(f"DEBUG: is_blocked='{is_blocked}' - EXIT!")
+ if block_level is None:
+ database.cursor.execute(
+ "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? LIMIT 1",
+ [
+ blocker,
+ blocked
+ ]
+ )
+ else:
+ database.cursor.execute(
+ "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
+ [
+ blocker,
+ blocked,
+ block_level
+ ]
+ )
+
+ is_blocked = database.cursor.fetchone() is not None
+
+ logger.debug("is_blocked='%s' - EXIT!", is_blocked)
return is_blocked
-def add_instance(blocker: str, blocked: str, reason: str, block_level: str):
- # DEBUG: print("DEBUG: blocker,blocked,reason,block_level:", blocker, blocked, reason, block_level)
- if not isinstance(blocker, str):
- raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
- elif blocker == "":
- raise ValueError("Parameter 'blocker' is empty")
- elif not validators.domain(blocker.split("/")[0]):
- raise ValueError(f"Bad blocker='{blocker}'")
- elif not isinstance(blocked, str):
- raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
- elif blocked == "":
- raise ValueError("Parameter 'blocked' is empty")
- elif not isinstance(block_level, str):
+def add(blocker: str, blocked: str, reason: str, block_level: str):
+ logger.debug("blocker='%s',blocked='%s',reason='%s',block_level='%s' - CALLED!", blocker, blocked, reason, block_level)
+ domain_helper.raise_on(blocker)
+ domain_helper.raise_on(blocked)
+
+ if not isinstance(block_level, str):
raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
elif block_level == "":
raise ValueError("Parameter 'block_level' is empty")
- elif not validators.domain(blocked.split("/")[0]):
- raise ValueError(f"Bad blocked='{blocked}'")
+ elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
+ raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
elif blacklist.is_blacklisted(blocker):
raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
elif blacklist.is_blacklisted(blocked):
raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
- elif block_level == "accept":
- raise ValueError("Accepted domains are not wanted here")
+ elif reason is not None and reason == "":
+ raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' has empty (not 'None') block reason set")
if reason is not None:
# Maybe needs cleaning
+ logger.debug("reason='%s' - BEFORE!")
reason = tidyup.reason(reason)
+ logger.debug("reason='%s' - AFTER!")
- print(f"INFO: New block: blocker='{blocker}',blocked='{blocked}',reason='{reason}',block_level='{block_level}'")
+ logger.info("New block: blocker='%s',blocked='%s',reason='%s',block_level='%s'", blocker, blocked, reason, block_level)
- fba.cursor.execute(
+ database.cursor.execute(
"INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?)",
[
blocker,
time.time()
])
- # DEBUG: print("DEBUG: EXIT!")
+ logger.debug("EXIT!")
+
+def valid(value: str, column: str) -> bool:
+ logger.debug("value='%s' - CALLED!", value)
+
+ if not isinstance(value, str):
+ raise ValueError(f"Parameter value[]='{type(value)}' is not of type 'str'")
+ elif value == "":
+ raise ValueError("Parameter 'value' is empty")
+ elif not isinstance(column, str):
+ raise ValueError(f"Parameter column[]='{type(column)}' is not of type 'str'")
+ elif column == "":
+ raise ValueError("Parameter 'column' is empty")
+
+ # Query database
+ database.cursor.execute(
+ f"SELECT {column} FROM blocks WHERE {column} = ? LIMIT 1", [value]
+ )
+
+ is_valid = database.cursor.fetchone() is not None
+
+ logger.debug("is_valid='%s' - EXIT!", is_valid)
+ return is_valid
+
+def translate_idnas(rows: list, column: str):
+ logger.debug("rows[]='%s' - CALLED!", type(rows))
+
+ if not isinstance(rows, list):
+ raise ValueError(f"rows[]='{type(rows)}' is not of type 'list'")
+ elif len(rows) == 0:
+ raise ValueError("Parameter 'rows' is an empty list")
+ elif not isinstance(column, str):
+ raise ValueError(f"column='{type(column)}' is not of type 'str'")
+ elif column not in ["blocker", "blocked"]:
+ raise ValueError(f"column='{column}' is not supported")
+
+ logger.info("Checking/converting %d domain names ...", len(rows))
+ for row in rows:
+ logger.debug("row[]='%s'", type(row))
+
+ translated = row[column].encode("idna").decode("utf-8")
+ logger.debug("translated='%s',row[%s]='%s'", translated, column, row[column])
+
+ if translated != row[column]:
+ logger.info("Translated row[%s]='%s' to '%s'", column, row[column], translated)
+ database.cursor.execute(f"UPDATE blocks SET {column} = ? WHERE {column} = ?", [translated, row[column]])
+
+ logger.debug("Invoking commit() ...")
+ database.connection.commit()
+
+ logger.debug("EXIT!")
+
+def alias_block_level(block_level: str) -> str:
+ logger.debug("block_level='%s' - CALLED!", block_level)
+
+ if not isinstance(block_level, str):
+ raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
+ elif block_level in ["accept", "accepted"]:
+ raise ValueError(f"Parameter block_level='{block_level}' is not accepted but function was invoked")
+ elif block_level == "silence":
+ logger.debug("Block level 'silence' has been changed to 'silenced'")
+ block_level = "silenced"
+ elif block_level == "suspend":
+ logger.debug("Block level 'suspend' has been changed to 'suspended'")
+ block_level = "suspended"
+ elif block_level == "nsfw":
+ logger.debug("Block level 'nsfw' has been changed to 'media_nsfw'")
+ block_level = "media_nsfw"
+ elif block_level == "quarantined_instances":
+ logger.debug("Block level 'quarantined_instances' has been changed to 'quarantined'")
+ block_level = "quarantined"
+
+ logger.debug("block_level='%s' - EXIT!", block_level)
+ return block_level
projects / fba.git / blobdiff