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 not is_instance_blocked(blocker, blocked, block_level):
44 raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' is not registered but function is invoked")
46 logger.debug("Updating block reason='%s',blocker='%s',blocked='%s',block_level='%s'", reason, blocker, blocked, block_level)
47 database.cursor.execute(
48 "UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND (reason IS NULL OR reason = '') LIMIT 1",
59 def update_last_seen(blocker: str, blocked: str, block_level: str):
60 logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
61 domain_helper.raise_on(blocker)
62 domain_helper.raise_on(blocked)
64 if not isinstance(block_level, str):
65 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
66 elif block_level == "":
67 raise ValueError("Parameter 'block_level' is empty")
68 elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
69 raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
70 elif blacklist.is_blacklisted(blocker):
71 raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
72 elif blacklist.is_blacklisted(blocked):
73 raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
74 elif not is_instance_blocked(blocker, blocked, block_level):
75 raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' is not registered but function is invoked")
77 database.cursor.execute(
78 "UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
88 def is_instance_blocked(blocker: str, blocked: str, block_level: str = None) -> bool:
89 logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
90 domain_helper.raise_on(blocker)
91 domain_helper.raise_on(blocked)
93 if not isinstance(block_level, str) and block_level is not None:
94 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
95 elif block_level == "":
96 raise ValueError("Parameter 'block_level' is empty")
97 elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
98 raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
99 elif blacklist.is_blacklisted(blocker):
100 raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
101 elif blacklist.is_blacklisted(blocked):
102 raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
104 if block_level is None:
105 database.cursor.execute(
106 "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? LIMIT 1",
113 database.cursor.execute(
114 "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
122 is_blocked = database.cursor.fetchone() is not None
124 logger.debug("is_blocked='%s' - EXIT!", is_blocked)
127 def add(blocker: str, blocked: str, reason: str, block_level: str):
128 logger.debug("blocker='%s',blocked='%s',reason='%s',block_level='%s' - CALLED!", blocker, blocked, reason, block_level)
129 domain_helper.raise_on(blocker)
130 domain_helper.raise_on(blocked)
132 if not isinstance(block_level, str):
133 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
134 elif block_level == "":
135 raise ValueError("Parameter 'block_level' is empty")
136 elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
137 raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
138 elif blacklist.is_blacklisted(blocker):
139 raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
140 elif blacklist.is_blacklisted(blocked):
141 raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
142 elif reason is not None and reason == "":
143 raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' has empty (not 'None') block reason set")
145 if reason is not None:
146 # Maybe needs cleaning
147 logger.debug("reason='%s' - BEFORE!")
148 reason = tidyup.reason(reason)
149 logger.debug("reason='%s' - AFTER!")
151 logger.info("New block: blocker='%s',blocked='%s',reason='%s',block_level='%s'", blocker, blocked, reason, block_level)
153 database.cursor.execute(
154 "INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?)",
164 logger.debug("EXIT!")
166 def valid(value: str, column: str) -> bool:
167 logger.debug("value='%s' - CALLED!", value)
169 if not isinstance(value, str):
170 raise ValueError(f"Parameter value[]='{type(value)}' is not of type 'str'")
172 raise ValueError("Parameter 'value' is empty")
173 elif not isinstance(column, str):
174 raise ValueError(f"Parameter column[]='{type(column)}' is not of type 'str'")
176 raise ValueError("Parameter 'column' is empty")
179 database.cursor.execute(
180 f"SELECT {column} FROM blocks WHERE {column} = ? LIMIT 1", [value]
183 is_valid = database.cursor.fetchone() is not None
185 logger.debug("is_valid='%s' - EXIT!", is_valid)
188 def translate_idnas(rows: list, column: str):
189 logger.debug("rows[]='%s' - CALLED!", type(rows))
191 if not isinstance(rows, list):
192 raise ValueError(f"rows[]='{type(rows)}' is not of type 'list'")
194 raise ValueError("Parameter 'rows' is an empty list")
195 elif not isinstance(column, str):
196 raise ValueError(f"column='{type(column)}' is not of type 'str'")
197 elif column not in ["blocker", "blocked"]:
198 raise ValueError(f"column='{column}' is not supported")
200 logger.info("Checking/converting %d domain names ...", len(rows))
202 logger.debug("row[]='%s'", type(row))
204 translated = row[column].encode("idna").decode("utf-8")
205 logger.debug("translated='%s',row[%s]='%s'", translated, column, row[column])
207 if translated != row[column]:
208 logger.info("Translated row[%s]='%s' to '%s'", column, row[column], translated)
209 database.cursor.execute(f"UPDATE blocks SET {column} = ? WHERE {column} = ?", [translated, row[column]])
211 logger.debug("Invoking commit() ...")
212 database.connection.commit()
214 logger.debug("EXIT!")
216 def alias_block_level(block_level: str) -> str:
217 logger.debug("block_level='%s' - CALLED!", block_level)
219 if not isinstance(block_level, str):
220 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
221 elif block_level in ["accept", "accepted"]:
222 raise ValueError(f"Parameter block_level='{block_level}' is not accepted but function was invoked")
223 elif block_level == "silence":
224 logger.debug("Block level 'silence' has been changed to 'silenced'")
225 block_level = "silenced"
226 elif block_level == "suspend":
227 logger.debug("Block level 'suspend' has been changed to 'suspended'")
228 block_level = "suspended"
229 elif block_level == "nsfw":
230 logger.debug("Block level 'nsfw' has been changed to 'media_nsfw'")
231 block_level = "media_nsfw"
232 elif block_level == "quarantined_instances":
233 logger.debug("Block level 'quarantined_instances' has been changed to 'quarantined'")
234 block_level = "quarantined"
236 logger.debug("block_level='%s' - EXIT!", block_level)