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")
143 if reason is not None:
144 # Maybe needs cleaning
145 reason = tidyup.reason(reason)
147 logger.info("New block: blocker='%s',blocked='%s',reason='%s',block_level='%s'", blocker, blocked, reason, block_level)
149 database.cursor.execute(
150 "INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?)",
160 logger.debug("EXIT!")
162 def valid(value: str, column: str) -> bool:
163 logger.debug("value='%s' - CALLED!", value)
165 if not isinstance(value, str):
166 raise ValueError(f"Parameter value[]='{type(value)}' is not of type 'str'")
168 raise ValueError("Parameter 'value' is empty")
169 elif not isinstance(column, str):
170 raise ValueError(f"Parameter column[]='{type(column)}' is not of type 'str'")
172 raise ValueError("Parameter 'column' is empty")
175 database.cursor.execute(
176 f"SELECT {column} FROM blocks WHERE {column} = ? LIMIT 1", [value]
179 is_valid = database.cursor.fetchone() is not None
181 logger.debug("is_valid='%s' - EXIT!", is_valid)
184 def translate_idnas(rows: list, column: str):
185 logger.debug("rows[]='%s' - CALLED!", type(rows))
187 if not isinstance(rows, list):
188 raise ValueError(f"rows[]='{type(rows)}' is not of type 'list'")
190 raise ValueError("Parameter 'rows' is an empty list")
191 elif not isinstance(column, str):
192 raise ValueError(f"column='{type(column)}' is not of type 'str'")
193 elif column not in ["blocker", "blocked"]:
194 raise ValueError(f"column='{column}' is not supported")
196 logger.info("Checking/converting %d domain names ...", len(rows))
198 logger.debug("row[]='%s'", type(row))
200 translated = row[column].encode("idna").decode("utf-8")
201 logger.debug("translated='%s',row[%s]='%s'", translated, column, row[column])
203 if translated != row[column]:
204 logger.info("Translated row[%s]='%s' to '%s'", column, row[column], translated)
205 database.cursor.execute(f"UPDATE blocks SET {column} = ? WHERE {column} = ?", [translated, row[column]])
207 logger.debug("Invoking commit() ...")
208 database.connection.commit()
210 logger.debug("EXIT!")
212 def alias_block_level(block_level: str) -> str:
213 logger.debug("block_level='%s' - CALLED!", block_level)
215 if not isinstance(block_level, str):
216 raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
217 elif block_level in ["accept", "accepted"]:
218 raise ValueError(f"Parameter block_level='{block_level}' is not accepted but function was invoked")
219 elif block_level == "silence":
220 logger.debug("Block level 'silence' has been changed to 'silenced'")
221 block_level = "silenced"
222 elif block_level == "suspend":
223 logger.debug("Block level 'suspend' has been changed to 'suspended'")
224 block_level = "suspended"
225 elif block_level == "nsfw":
226 logger.debug("Block level 'nsfw' has been changed to 'media_nsfw'")
227 block_level = "media_nsfw"
228 elif block_level == "quarantined_instances":
229 logger.debug("Block level 'quarantined_instances' has been changed to 'quarantined'")
230 block_level = "quarantined"
232 logger.debug("block_level='%s' - EXIT!", block_level)