From: Roland Häder Date: Sun, 11 Jun 2023 12:21:57 +0000 (+0200) Subject: Continued: X-Git-Url: https://git.mxchange.org/?a=commitdiff_plain;h=481de3ec49efdfdbff0b442bb0d7bd8c26328e39;p=fba.git Continued: - fetch[_wellknown]_nodeinfo() should return same dict with at least 'status_code' and 'json' on success and 'error_message' on failure --- diff --git a/fba/federation.py b/fba/federation.py index 5b4c4ba..abee665 100644 --- a/fba/federation.py +++ b/fba/federation.py @@ -122,14 +122,15 @@ def fetch_peers(domain: str, software: str) -> list: # Init peers variable peers = list() + # No CSRF by default, you don't have to add network.api_headers by yourself here headers = tuple() - # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") try: - headers = csrf.determine(domain, dict()) + # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") + headers = csrf.determine(domain, dict()) except network.exceptions as exception: - print(f"WARNING: Exception '{type(exception)}' during checking CSRF - EXIT!") - return + print(f"WARNING: Exception '{type(exception)}' during checking CSRF (fetch_peers,{__name__}) - EXIT!") + return peers # DEBUG: print(f"DEBUG: Fetching peers from '{domain}',software='{software}' ...") data = network.get_json_api( @@ -173,7 +174,7 @@ def fetch_peers(domain: str, software: str) -> list: # DEBUG: print("DEBUG: Returning peers[]:", type(peers)) return peers -def fetch_nodeinfo(domain: str, path: str = None) -> list: +def fetch_nodeinfo(domain: str, path: str = None) -> dict: # DEBUG: print(f"DEBUG: domain='{domain}',path={path} - CALLED!") if not isinstance(domain, str): raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'") @@ -185,19 +186,20 @@ def fetch_nodeinfo(domain: str, path: str = None) -> list: # DEBUG: print(f"DEBUG: Fetching nodeinfo from domain='{domain}' ...") nodeinfo = fetch_wellknown_nodeinfo(domain) - # DEBUG: print(f"DEBUG: nodeinfo({len(nodeinfo)})={nodeinfo}") - if len(nodeinfo) > 0: - # DEBUG: print("DEBUG: nodeinfo()={len(nodeinfo))} - EXIT!") + # DEBUG: print(f"DEBUG: nodeinfo[{type(nodeinfo)}]='{nodeinfo}'") + if "error_message" in nodeinfo: + print(f"WARNING: Error during fetching nodeinfo: '{nodeinfo['error_message']}' - EXIT!") return nodeinfo + # No CSRF by default, you don't have to add network.api_headers by yourself here headers = tuple() - # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") try: - headers = csrf.determine(domain, dict()) + # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") + headers = csrf.determine(domain, dict()) except network.exceptions as exception: - print(f"WARNING: Exception '{type(exception)}' during checking CSRF - EXIT!") - return + print(f"WARNING: Exception '{type(exception)}' during checking CSRF (nodeinfo,{__name__}) - EXIT!") + return dict() request_paths = [ "/nodeinfo/2.1.json", @@ -209,6 +211,7 @@ def fetch_nodeinfo(domain: str, path: str = None) -> list: ] for request in request_paths: + # DEBUG: print(f"DEBUG: path[{type(path)}]='{path}',request='{request'}") if path is not None and path != "" and path != request: # DEBUG: print(f"DEBUG: path='{path}' does not match request='{request}' - SKIPPED!") continue @@ -233,21 +236,25 @@ def fetch_nodeinfo(domain: str, path: str = None) -> list: # DEBUG: print(f"DEBUG: data()={len(data)} - EXIT!") return data -def fetch_wellknown_nodeinfo(domain: str) -> list: +def fetch_wellknown_nodeinfo(domain: str) -> dict: # DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!") if not isinstance(domain, str): raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'") elif domain == "": raise ValueError("Parameter 'domain' is empty") + # No CSRF by default, you don't have to add network.api_headers by yourself here headers = tuple() - # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") try: - headers = csrf.determine(domain, dict()) + # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") + headers = csrf.determine(domain, dict()) except network.exceptions as exception: - print(f"WARNING: Exception '{type(exception)}' during checking CSRF - EXIT!") - return + print(f"WARNING: Exception '{type(exception)}' during checking CSRF (fetch_wellknown,{__name__}) - EXIT!") + return { + "status_code" : 500, + "error_message": type(exception) + } # DEBUG: print("DEBUG: Fetching .well-known info for domain:", domain) data = network.get_json_api( @@ -365,7 +372,7 @@ def determine_software(domain: str, path: str = None) -> str: # DEBUG: print(f"DEBUG: Fetching nodeinfo from '{domain}' ...") data = fetch_nodeinfo(domain, path) - # DEBUG: print("DEBUG: data[]:", type(data)) + # DEBUG: print(f"DEBUG: data[{type(data)}]='{data}'") if "error_message" in data: # DEBUG: print("DEBUG: Could not determine software type:", domain) return fetch_generator_from_path(domain) diff --git a/fba/networks/lemmy.py b/fba/networks/lemmy.py index 82078be..cba46b1 100644 --- a/fba/networks/lemmy.py +++ b/fba/networks/lemmy.py @@ -27,16 +27,18 @@ def fetch_peers(domain: str) -> list: elif domain == "": raise ValueError("Parameter 'domain' is empty") + peers = list() + + # No CSRF by default, you don't have to add network.api_headers by yourself here headers = tuple() - # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") try: - headers = csrf.determine(domain, dict()) + # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") + headers = csrf.determine(domain, dict()) except network.exceptions as exception: - print(f"WARNING: Exception '{type(exception)}' during checking CSRF - EXIT!") - return + print(f"WARNING: Exception '{type(exception)}' during checking CSRF (fetch_peers,{__name__}) - EXIT!") + return peers - peers = list() try: # DEBUG: print(f"DEBUG: domain='{domain}' is Lemmy, fetching JSON ...") data = network.get_json_api( diff --git a/fba/networks/mastodon.py b/fba/networks/mastodon.py index 7f6a0d5..2019467 100644 --- a/fba/networks/mastodon.py +++ b/fba/networks/mastodon.py @@ -131,13 +131,14 @@ def fetch_blocks(domain: str, origin: str, nodeinfo_url: str): elif nodeinfo_url == "": raise ValueError("Parameter 'nodeinfo_url' is empty") + # No CSRF by default, you don't have to add network.api_headers by yourself here headers = tuple() - # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") try: - headers = csrf.determine(domain, dict()) + # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") + headers = csrf.determine(domain, dict()) except network.exceptions as exception: - print(f"WARNING: Exception '{type(exception)}' during checking CSRF - EXIT!") + print(f"WARNING: Exception '{type(exception)}' during checking CSRF (fetch_blocks,{__name__}) - EXIT!") return try: diff --git a/fba/networks/misskey.py b/fba/networks/misskey.py index 6b24628..0ae00d7 100644 --- a/fba/networks/misskey.py +++ b/fba/networks/misskey.py @@ -38,14 +38,16 @@ def fetch_peers(domain: str) -> list: peers = list() offset = 0 step = config.get("misskey_limit") + + # No CSRF by default, you don't have to add network.api_headers by yourself here headers = tuple() - # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") try: - headers = csrf.determine(domain, dict()) + # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") + headers = csrf.determine(domain, dict()) except network.exceptions as exception: - print(f"WARNING: Exception '{type(exception)}' during checking CSRF - EXIT!") - return + print(f"WARNING: Exception '{type(exception)}' during checking CSRF (fetch_peers,{__name__}) - EXIT!") + return peers # iterating through all "suspended" (follow-only in its terminology) # instances page-by-page, since that troonware doesn't support @@ -139,14 +141,16 @@ def fetch_blocks(domain: str) -> dict: offset = 0 step = config.get("misskey_limit") + + # No CSRF by default, you don't have to add network.api_headers by yourself here headers = tuple() - # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") try: - headers = csrf.determine(domain, dict()) + # DEBUG: print(f"DEBUG: Checking CSRF for domain='{domain}'") + headers = csrf.determine(domain, dict()) except network.exceptions as exception: - print(f"WARNING: Exception '{type(exception)}' during checking CSRF - EXIT!") - return + print(f"WARNING: Exception '{type(exception)}' during checking CSRF (fetch_blocks,{__name__}) - EXIT!") + return blocklist # iterating through all "suspended" (follow-only in its terminology) # instances page-by-page since it doesn't support sending them all at once diff --git a/fba/networks/peertube.py b/fba/networks/peertube.py index 5eb7946..a493139 100644 --- a/fba/networks/peertube.py +++ b/fba/networks/peertube.py @@ -29,14 +29,16 @@ def fetch_peers(domain: str) -> list: print(f"DEBUG: domain='{domain}' is a PeerTube, fetching JSON ...") peers = list() start = 0 + + # No CSRF by default, you don't have to add network.api_headers by yourself here headers = tuple() - print(f"DEBUG: Checking CSRF for domain='{domain}'") try: - headers = csrf.determine(domain, dict()) + print(f"DEBUG: Checking CSRF for domain='{domain}'") + headers = csrf.determine(domain, dict()) except network.exceptions as exception: - print(f"WARNING: Exception '{type(exception)}' during checking CSRF - EXIT!") - return + print(f"WARNING: Exception '{type(exception)}' during checking CSRF (fetch_peers,{__name__}) - EXIT!") + return peers for mode in ["followers", "following"]: print(f"DEBUG: domain='{domain}',mode='{mode}'")