1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
2 # Copyright (C) 2023 Free Software Foundation
4 # This program is free software: you can redistribute it and/or modify
5 # it under the terms of the GNU Affero General Public License as published
6 # by the Free Software Foundation, either version 3 of the License, or
7 # (at your option) any later version.
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU Affero General Public License for more details.
14 # You should have received a copy of the GNU Affero General Public License
15 # along with this program. If not, see <https://www.gnu.org/licenses/>.
21 from fba import database
23 from fba.helpers import blacklist
24 from fba.helpers import domain as domain_helper
25 from fba.helpers import tidyup
27 logging.basicConfig(level=logging.INFO)
28 logger = logging.getLogger(__name__)
30 def update_reason(reason: str, blocker: str, blocked: str, block_level: str):
31 logger.debug("reason='%s',blocker='%s',blocked='%s',block_level='%s' - CALLED!", reason, blocker, blocked, block_level)
32 domain_helper.raise_on(blocker)
33 domain_helper.raise_on(blocked)
35 if not isinstance(reason, str) and reason is not None:
36 raise ValueError(f"Parameter reason[]='{type(reason)}' is not of type 'str'")
37 elif not isinstance(block_level, str):
38 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
39 elif block_level == "":
40 raise ValueError("Parameter 'block_level' is empty")
41 elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
42 raise ValueError(f"block_level='{block_level}' is not wanted.")
43 elif blacklist.is_blacklisted(blocker):
44 raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
45 elif blacklist.is_blacklisted(blocked):
46 raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
47 elif not is_instance_blocked(blocker, blocked, block_level):
48 raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' is not registered but function is invoked")
50 logger.debug("Updating block reason='%s',blocker='%s',blocked='%s',block_level='%s'", reason, blocker, blocked, block_level)
51 database.cursor.execute(
52 "UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND (reason IS NULL OR reason = '') LIMIT 1",
63 def update_last_seen(blocker: str, blocked: str, block_level: str):
64 logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
65 domain_helper.raise_on(blocker)
66 domain_helper.raise_on(blocked)
68 if not isinstance(block_level, str):
69 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
70 elif block_level == "":
71 raise ValueError("Parameter 'block_level' is empty")
72 elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
73 raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
74 elif blacklist.is_blacklisted(blocker):
75 raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
76 elif blacklist.is_blacklisted(blocked):
77 raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
78 elif not is_instance_blocked(blocker, blocked, block_level):
79 raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' is not registered but function is invoked")
81 database.cursor.execute(
82 "UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
92 def is_instance_blocked(blocker: str, blocked: str, block_level: str = None) -> bool:
93 logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
94 domain_helper.raise_on(blocker)
95 domain_helper.raise_on(blocked)
97 if not isinstance(block_level, str) and block_level is not None:
98 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
99 elif block_level == "":
100 raise ValueError("Parameter 'block_level' is empty")
101 elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
102 raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
103 elif blacklist.is_blacklisted(blocker):
104 raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
105 elif blacklist.is_blacklisted(blocked):
106 raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
108 if block_level is None:
109 database.cursor.execute(
110 "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? LIMIT 1",
117 database.cursor.execute(
118 "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
126 is_blocked = database.cursor.fetchone() is not None
128 logger.debug("is_blocked='%s' - EXIT!", is_blocked)
131 def add(blocker: str, blocked: str, reason: str, block_level: str):
132 logger.debug("blocker='%s',blocked='%s',reason='%s',block_level='%s' - CALLED!", blocker, blocked, reason, block_level)
133 domain_helper.raise_on(blocker)
134 domain_helper.raise_on(blocked)
136 if not isinstance(block_level, str):
137 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
138 elif block_level == "":
139 raise ValueError("Parameter 'block_level' is empty")
140 elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
141 raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
142 elif blacklist.is_blacklisted(blocker):
143 raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
144 elif blacklist.is_blacklisted(blocked):
145 raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
146 elif reason is not None and reason == "":
147 raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' has empty (not 'None') block reason set")
149 if reason is not None:
150 # Maybe needs cleaning
151 logger.debug("reason='%s' - BEFORE!")
152 reason = tidyup.reason(reason)
153 logger.debug("reason='%s' - AFTER!")
155 logger.info("New block: blocker='%s',blocked='%s',reason='%s',block_level='%s'", blocker, blocked, reason, block_level)
157 database.cursor.execute(
158 "INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?)",
168 logger.debug("EXIT!")
170 def valid(value: str, column: str) -> bool:
171 logger.debug("value='%s' - CALLED!", value)
173 if not isinstance(value, str):
174 raise ValueError(f"Parameter value[]='{type(value)}' is not of type 'str'")
176 raise ValueError("Parameter 'value' is empty")
177 elif not isinstance(column, str):
178 raise ValueError(f"Parameter column[]='{type(column)}' is not of type 'str'")
180 raise ValueError("Parameter 'column' is empty")
183 database.cursor.execute(
184 f"SELECT {column} FROM blocks WHERE {column} = ? LIMIT 1", [value]
187 is_valid = database.cursor.fetchone() is not None
189 logger.debug("is_valid='%s' - EXIT!", is_valid)
192 def translate_idnas(rows: list, column: str):
193 logger.debug("rows[]='%s' - CALLED!", type(rows))
195 if not isinstance(rows, list):
196 raise ValueError(f"rows[]='{type(rows)}' is not of type 'list'")
198 raise ValueError("Parameter 'rows' is an empty list")
199 elif not isinstance(column, str):
200 raise ValueError(f"column='{type(column)}' is not of type 'str'")
201 elif column not in ["blocker", "blocked"]:
202 raise ValueError(f"column='{column}' is not supported")
204 logger.info("Checking/converting %d domain names ...", len(rows))
206 logger.debug("row[]='%s'", type(row))
208 translated = row[column].encode("idna").decode("utf-8")
209 logger.debug("translated='%s',row[%s]='%s'", translated, column, row[column])
211 if translated != row[column]:
212 logger.info("Translated row[%s]='%s' to '%s'", column, row[column], translated)
213 database.cursor.execute(f"UPDATE blocks SET {column} = ? WHERE {column} = ?", [translated, row[column]])
215 logger.debug("Invoking commit() ...")
216 database.connection.commit()
218 logger.debug("EXIT!")
220 def alias_block_level(block_level: str) -> str:
221 logger.debug("block_level='%s' - CALLED!", block_level)
223 if not isinstance(block_level, str):
224 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
225 elif block_level in ["accept", "accepted"]:
226 raise ValueError(f"Parameter block_level='{block_level}' is not accepted but function was invoked")
227 elif block_level == "silence":
228 logger.debug("Block level 'silence' has been changed to 'silenced'")
229 block_level = "silenced"
230 elif block_level == "suspend":
231 logger.debug("Block level 'suspend' has been changed to 'suspended'")
232 block_level = "suspended"
233 elif block_level == "nsfw":
234 logger.debug("Block level 'nsfw' has been changed to 'media_nsfw'")
235 block_level = "media_nsfw"
236 elif block_level == "quarantined_instances":
237 logger.debug("Block level 'quarantined_instances' has been changed to 'quarantined'")
238 block_level = "quarantined"
240 logger.debug("block_level='%s' - EXIT!", block_level)