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/>.
17 from fba import config
19 from fba import federation
20 from fba import network
22 from fba.models import instances
24 def fetch_peers(domain: str) -> list:
25 # DEBUG: print(f"DEBUG: domain({len(domain)})='{domain}',software='lemmy' - CALLED!")
26 if not isinstance(domain, str):
27 raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
29 raise ValueError("Parameter 'domain' is empty")
33 # No CSRF by default, you don't have to add network.api_headers by yourself here
37 # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'")
38 headers = csrf.determine(domain, dict())
39 except network.exceptions as exception:
40 print(f"WARNING: Exception '{type(exception)}' during checking CSRF (fetch_peers,{__name__}) - EXIT!")
41 instances.set_last_error(domain, exception)
45 # DEBUG: print(f"DEBUG: domain='{domain}' is Lemmy, fetching JSON ...")
46 data = network.get_json_api(
50 (config.get("connection_timeout"), config.get("read_timeout"))
53 # DEBUG: print(f"DEBUG: data[]='{type(data)}'")
54 if "error_message" in data:
55 print("WARNING: Could not reach any JSON API:", domain)
56 instances.set_last_error(domain, data)
57 elif "federated_instances" in data["json"]:
58 # DEBUG: print(f"DEBUG: Found federated_instances for domain='{domain}'")
59 peers = peers + federation.add_peers(data["json"]["federated_instances"])
60 # DEBUG: print("DEBUG: Added instance(s) to peers")
62 print("WARNING: JSON response does not contain 'federated_instances':", domain)
63 instances.set_last_error(domain, data)
65 except BaseException as exception:
66 print(f"WARNING: Exception during fetching JSON: domain='{domain}',exception[{type(exception)}]:'{str(exception)}'")
68 # DEBUG: print(f"DEBUG: Adding '{len(peers)}' for domain='{domain}'")
69 instances.set_total_peers(domain, peers)
71 # DEBUG: print("DEBUG: Returning peers[]:", type(peers))