fba/network.py

   1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
   2 # Copyright (C) 2023 Free Software Foundation
   3 #
   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.
   8 #
   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.
  13 #
  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/>.
  16
  17 import json
  18 import reqto
  19 import requests
  20
  21 from fba import config
  22 from fba import csrf
  23 from fba import instances
  24
  25 # HTTP headers for non-API requests
  26 web_headers = {
  27     "User-Agent": config.get("useragent"),
  28 }
  29
  30 # HTTP headers for API requests
  31 api_headers = {
  32     "User-Agent"  : config.get("useragent"),
  33     "Content-Type": "application/json",
  34 }
  35
  36 def post_json_api(domain: str, path: str, parameter: str, extra_headers: dict = {}) -> dict:
  37     print(f"DEBUG: domain='{domain}',path='{path}',parameter='{parameter}',extra_headers()={len(extra_headers)} - CALLED!")
  38     if not isinstance(domain, str):
  39         raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
  40     elif domain == "":
  41         raise ValueError("Parameter 'domain' is empty")
  42     elif not isinstance(path, str):
  43         raise ValueError(f"path[]={type(path)} is not 'str'")
  44     elif path == "":
  45         raise ValueError("Parameter 'path' cannot be empty")
  46     elif not isinstance(parameter, str):
  47         raise ValueError(f"parameter[]={type(parameter)} is not 'str'")
  48
  49     print(f"DEBUG: Determining if CSRF header needs to be sent for domain='{domain}' ...")
  50     headers = csrf.determine(domain, {**api_headers, **extra_headers})
  51
  52     data = {}
  53
  54     try:
  55         print(f"DEBUG: Sending POST to domain='{domain}',path='{path}',parameter='{parameter}',extra_headers({len(extra_headers)})={extra_headers}")
  56         response = reqto.post(
  57             f"https://{domain}{path}",
  58             data=parameter,
  59             headers=headers,
  60             timeout=(config.get("connection_timeout"), config.get("read_timeout"))
  61         )
  62
  63         data = json_from_response(response)
  64         print(f"DEBUG: response.ok={response.ok},response.status_code={response.status_code},data[]='{type(data)}'")
  65         if not response.ok or response.status_code >= 400:
  66             print(f"WARNING: Cannot query JSON API: domain='{domain}',path='{path}',parameter()={len(parameter)},response.status_code='{response.status_code}',data[]='{type(data)}'")
  67             instances.update_last_error(domain, response)
  68
  69     except BaseException as exception:
  70         print(f"WARNING: Some error during post(): domain='{domain}',path='{path}',parameter()={len(parameter)},exception[{type(exception)}]:'{str(exception)}'")
  71
  72     print(f"DEBUG: Returning data({len(data)})=[]:{type(data)}")
  73     return data
  74
  75 def send_bot_post(domain: str, blocklist: dict):
  76     print(f"DEBUG: domain={domain},blocklist()={len(blocklist)} - CALLED!")
  77     if not isinstance(domain, str):
  78         raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
  79     elif domain == "":
  80         raise ValueError("Parameter 'domain' is empty")
  81     elif not isinstance(blocklist, dict):
  82         raise ValueError(f"Parameter blocklist[]='{type(blocklist)}' is not 'dict'")
  83
  84     message = f"{domain} has blocked the following instances:\n\n"
  85     truncated = False
  86
  87     if len(blocklist) > 20:
  88         truncated = True
  89         blocklist = blocklist[0 : 19]
  90
  91     print(f"DEBUG: blocklist()={len(blocklist)}")
  92     for block in blocklist:
  93         print(f"DEBUG: block['{type(block)}']={block}")
  94         if block["reason"] is None or block["reason"] == '':
  95             message = message + block["blocked"] + " with unspecified reason\n"
  96         else:
  97             if len(block["reason"]) > 420:
  98                 block["reason"] = block["reason"][0:419] + "[…]"
  99
 100             message = message + block["blocked"] + ' for "' + block["reason"].replace("@", "@\u200b") + '"\n'
 101
 102     if truncated:
 103         message = message + "(the list has been truncated to the first 20 entries)"
 104
 105     botheaders = {**api_headers, **{"Authorization": "Bearer " + config.get("bot_token")}}
 106
 107     req = reqto.post(
 108         f"{config.get('bot_instance')}/api/v1/statuses",
 109         data={
 110             "status"      : message,
 111             "visibility"  : config.get('bot_visibility'),
 112             "content_type": "text/plain"
 113         },
 114         headers=botheaders,
 115         timeout=10
 116     ).json()
 117
 118     return True
 119
 120 def fetch_response(domain: str, path: str, headers: dict, timeout: list) -> requests.models.Response:
 121     print(f"DEBUG: domain='{domain}',path='{path}',headers()={len(headers)},timeout={timeout} - CALLED!")
 122     if not isinstance(domain, str):
 123         raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
 124     elif domain == "":
 125         raise ValueError("Parameter 'domain' is empty")
 126     elif not isinstance(path, str):
 127         raise ValueError(f"Parameter path[]='{type(path)}' is not 'str'")
 128     elif path == "":
 129         raise ValueError("Parameter 'path' is empty")
 130
 131     print(f"DEBUG: Determining if CSRF header needs to be sent for domain='{domain}',headers()='{len(headers)}' ...")
 132     headers = csrf.determine(domain, headers)
 133
 134     try:
 135         print(f"DEBUG: Sending GET request to '{domain}{path}' ...")
 136         response = reqto.get(
 137             f"https://{domain}{path}",
 138             headers=headers,
 139             timeout=timeout
 140         )
 141
 142     except requests.exceptions.ConnectionError as exception:
 143         print(f"DEBUG: Fetching '{path}' from '{domain}' failed. exception[{type(exception)}]='{str(exception)}'")
 144         instances.update_last_error(domain, exception)
 145         raise exception
 146
 147     print(f"DEBUG: response[]='{type(response)}' - EXXIT!")
 148     return response
 149
 150 def json_from_response(response: requests.models.Response) -> list:
 151     print(f"DEBUG: response[]={type(response)} - CALLED!")
 152     if not isinstance(response, requests.models.Response):
 153         raise ValueError(f"Parameter response[]='{type(response)}' is not type of 'Response'")
 154
 155     data = list()
 156     if response.text.strip() != "":
 157         print(f"DEBUG: response.text()={len(response.text)} is not empty, invoking response.json() ...")
 158         try:
 159             data = response.json()
 160         except json.decoder.JSONDecodeError:
 161             pass
 162
 163     print(f"DEBUG: data[]={type(data)} - EXIT!")
 164     return data