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 config
23 from fba import instances
25 # HTTP headers for non-API requests
27 "User-Agent": config.get("useragent"),
30 # HTTP headers for API requests
32 "User-Agent" : config.get("useragent"),
33 "Content-Type": "application/json",
36 def post_json_api(domain: str, path: str, data: str, headers: dict = {}) -> dict:
37 # DEBUG: print(f"DEBUG: domain='{domain}',path='{path}',data='{data}',headers()={len(headers)} - CALLED!")
38 if not isinstance(domain, str):
39 raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
41 raise ValueError("Parameter 'domain' is empty")
42 elif not isinstance(path, str):
43 raise ValueError(f"path[]={type(path)} is not 'str'")
45 raise ValueError("Parameter 'path' cannot be empty")
46 elif not isinstance(data, str):
47 raise ValueError(f"data[]={type(data)} is not 'str'")
48 elif not isinstance(headers, dict):
49 raise ValueError(f"headers[]={type(headers)} is not 'list'")
56 # DEBUG: print(f"DEBUG: Sending POST to domain='{domain}',path='{path}',data='{data}',headers({len(headers)})={headers}")
57 response = reqto.post(
58 f"https://{domain}{path}",
60 headers={**api_headers, **headers},
61 timeout=(config.get("connection_timeout"), config.get("read_timeout"))
64 json_reply["json"] = json_from_response(response)
66 # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},json_reply[]='{type(json_reply)}'")
67 if not response.ok or response.status_code >= 400:
68 print(f"WARNING: Cannot query JSON API: domain='{domain}',path='{path}',data()={len(data)},response.status_code='{response.status_code}',json_reply[]='{type(json_reply)}'")
69 json_reply["status_code"] = response.status_code
70 json_reply["error_message"] = response.text
71 instances.update_last_error(domain, response)
73 except requests.exceptions.ConnectionError as exception:
74 # DEBUG: print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(exception)}]='{str(exception)}'")
75 json_reply["status_code"] = 999
76 json_reply["error_message"] = f"exception['{type(exception)}']='{str(exception)}'"
77 instances.update_last_error(domain, exception)
80 # DEBUG: print(f"DEBUG: Returning json_reply({len(json_reply)})=[]:{type(json_reply)}")
83 def fetch_api_url(url: str, timeout: tuple) -> dict:
84 # DEBUG: print(f"DEBUG: url='{url}',timeout()={len(timeout)} - CALLED!")
85 if not isinstance(url, str):
86 raise ValueError(f"Parameter url[]='{type(url)}' is not 'str'")
87 elif not isinstance(timeout, tuple):
88 raise ValueError(f"timeout[]={type(timeout)} is not 'tuple'")
95 # DEBUG: print(f"DEBUG: Fetching url='{url}' ...")
96 response = fba.fetch_url(url, api_headers, timeout)
98 json_reply["json"] = json_from_response(response)
100 # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},json_reply[]='{type(json_reply)}'")
101 if not response.ok or response.status_code >= 400:
102 print(f"WARNING: Cannot query JSON API: url='{url}',response.status_code='{response.status_code}',json_reply[]='{type(json_reply)}'")
103 json_reply["status_code"] = response.status_code
104 json_reply["error_message"] = response.text
106 except requests.exceptions.ConnectionError as exception:
107 # DEBUG: print(f"DEBUG: Fetching '{url}' failed. exception[{type(exception)}]='{str(exception)}'")
108 json_reply["status_code"] = 999
109 json_reply["error_message"] = f"exception['{type(exception)}']='{str(exception)}'"
112 # DEBUG: print(f"DEBUG: Returning json_reply({len(json_reply)})=[]:{type(json_reply)}")
115 def get_json_api(domain: str, path: str, headers: dict, timeout: tuple) -> dict:
116 # DEBUG: print(f"DEBUG: domain='{domain}',path='{path}',timeout()={len(timeout)} - CALLED!")
117 if not isinstance(domain, str):
118 raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
120 raise ValueError("Parameter 'domain' is empty")
121 elif not isinstance(path, str):
122 raise ValueError(f"path[]={type(path)} is not 'str'")
124 raise ValueError("Parameter 'path' cannot be empty")
125 elif not isinstance(headers, dict):
126 raise ValueError(f"headers[]={type(headers)} is not 'list'")
127 elif not isinstance(timeout, tuple):
128 raise ValueError(f"timeout[]={type(timeout)} is not 'tuple'")
135 # DEBUG: print(f"DEBUG: Sending GET to domain='{domain}',path='{path}',timeout({len(timeout)})={timeout}")
136 response = reqto.get(
137 f"https://{domain}{path}",
138 headers={**api_headers, **headers},
142 except requests.exceptions.ConnectionError as exception:
143 # DEBUG: print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(exception)}]='{str(exception)}'")
144 json_reply["status_code"] = 999
145 json_reply["error_message"] = f"exception['{type(exception)}']='{str(exception)}'"
146 instances.update_last_error(domain, exception)
149 json_reply["json"] = json_from_response(response)
151 # DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},json_reply[]='{type(json_reply)}'")
152 if not response.ok or response.status_code >= 400:
153 print(f"WARNING: Cannot query JSON API: domain='{domain}',path='{path}',response.status_code='{response.status_code}',json_reply[]='{type(json_reply)}'")
154 json_reply["status_code"] = response.status_code
155 json_reply["error_message"] = response.text
156 instances.update_last_error(domain, response)
158 # DEBUG: print(f"DEBUG: Returning json_reply({len(json_reply)})=[]:{type(json_reply)}")
161 def send_bot_post(domain: str, blocklist: dict):
162 # DEBUG: print(f"DEBUG: domain={domain},blocklist()={len(blocklist)} - CALLED!")
163 if not isinstance(domain, str):
164 raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
166 raise ValueError("Parameter 'domain' is empty")
167 elif not isinstance(blocklist, dict):
168 raise ValueError(f"Parameter blocklist[]='{type(blocklist)}' is not 'dict'")
170 message = f"{domain} has blocked the following instances:\n\n"
173 if len(blocklist) > 20:
175 blocklist = blocklist[0 : 19]
177 # DEBUG: print(f"DEBUG: blocklist()={len(blocklist)}")
178 for block in blocklist:
179 # DEBUG: print(f"DEBUG: block['{type(block)}']={block}")
180 if block["reason"] is None or block["reason"] == '':
181 message = message + block["blocked"] + " with unspecified reason\n"
183 if len(block["reason"]) > 420:
184 block["reason"] = block["reason"][0:419] + "[…]"
186 message = message + block["blocked"] + ' for "' + block["reason"].replace("@", "@\u200b") + '"\n'
189 message = message + "(the list has been truncated to the first 20 entries)"
191 botheaders = {**api_headers, **{"Authorization": "Bearer " + config.get("bot_token")}}
194 f"{config.get('bot_instance')}/api/v1/statuses",
197 "visibility" : config.get('bot_visibility'),
198 "content_type": "text/plain"
206 def fetch_response(domain: str, path: str, headers: dict, timeout: tuple) -> requests.models.Response:
207 # DEBUG: print(f"DEBUG: domain='{domain}',path='{path}',headers()={len(headers)},timeout={timeout} - CALLED!")
208 if not isinstance(domain, str):
209 raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
211 raise ValueError("Parameter 'domain' is empty")
212 elif not isinstance(path, str):
213 raise ValueError(f"Parameter path[]='{type(path)}' is not 'str'")
215 raise ValueError("Parameter 'path' is empty")
216 elif not isinstance(headers, dict):
217 raise ValueError(f"headers[]={type(headers)} is not 'dict'")
218 elif not isinstance(timeout, tuple):
219 raise ValueError(f"timeout[]={type(timeout)} is not 'tuple'")
222 # DEBUG: print(f"DEBUG: Sending GET request to '{domain}{path}' ...")
223 response = reqto.get(
224 f"https://{domain}{path}",
229 except requests.exceptions.ConnectionError as exception:
230 # DEBUG: print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(exception)}]='{str(exception)}'")
231 instances.update_last_error(domain, exception)
234 # DEBUG: print(f"DEBUG: response[]='{type(response)}' - EXXIT!")
237 def json_from_response(response: requests.models.Response) -> list:
238 # DEBUG: print(f"DEBUG: response[]={type(response)} - CALLED!")
239 if not isinstance(response, requests.models.Response):
240 raise ValueError(f"Parameter response[]='{type(response)}' is not type of 'Response'")
243 if response.text.strip() != "":
244 # DEBUG: print(f"DEBUG: response.text()={len(response.text)} is not empty, invoking response.json() ...")
246 data = response.json()
247 except json.decoder.JSONDecodeError:
250 # DEBUG: print(f"DEBUG: data[]={type(data)} - EXIT!")