fba/networks/lemmy.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 logging
  18
  19 import bs4
  20
  21 from fba import csrf
  22 from fba import utils
  23
  24 from fba.helpers import config
  25 from fba.helpers import domain as domain_helper
  26 from fba.helpers import tidyup
  27
  28 from fba.http import federation
  29 from fba.http import network
  30
  31 from fba.models import instances
  32
  33 logging.basicConfig(level=logging.INFO)
  34 logger = logging.getLogger(__name__)
  35
  36 def fetch_peers(domain: str) -> list:
  37     logger.debug("domain='%s' - CALLED!", domain)
  38     domain_helper.raise_on(domain)
  39
  40     peers = list()
  41
  42     # No CSRF by default, you don't have to add network.api_headers by yourself here
  43     headers = tuple()
  44
  45     try:
  46         logger.debug("Checking CSRF for domain='%s'", domain)
  47         headers = csrf.determine(domain, dict())
  48     except network.exceptions as exception:
  49         logger.warning("Exception '%s' during checking CSRF (fetch_peers,%s) - EXIT!", type(exception), __name__)
  50         instances.set_last_error(domain, exception)
  51         return list()
  52
  53     try:
  54         logger.debug("Fetching '/api/v3/site' from domain='%s' ...", domain)
  55         data = network.get_json_api(
  56             domain,
  57             "/api/v3/site",
  58             headers,
  59             (config.get("connection_timeout"), config.get("read_timeout"))
  60         )
  61
  62         logger.debug("data[]='%s'", type(data))
  63         if "error_message" in data:
  64             logger.warning("Could not reach any JSON API: domain='%s'", domain)
  65             instances.set_last_error(domain, data)
  66         elif "federated_instances" in data["json"] and isinstance(data["json"]["federated_instances"], dict):
  67             logger.debug("Found federated_instances for domain='%s'", domain)
  68             peers = peers + federation.add_peers(data["json"]["federated_instances"])
  69             instances.set_success(domain)
  70         else:
  71             logger.warning("JSON response does not contain 'federated_instances', domain='%s'", domain)
  72             instances.set_last_error(domain, data)
  73
  74     except network.exceptions as exception:
  75         logger.warning("Exception during fetching JSON: domain='%s',exception[%s]:'%s'", domain, type(exception), str(exception))
  76         instances.set_last_error(domain, exception)
  77
  78     logger.debug("Adding %d for domain='%s'", len(peers), domain)
  79     instances.set_total_peers(domain, peers)
  80
  81     logger.debug("peers()=%d - EXIT!", len(peers))
  82     return peers
  83
  84 def fetch_blocks(domain: str, nodeinfo_url: str) -> list:
  85     logger.debug("domain='%s,nodeinfo_url='%s' - CALLED!", domain, nodeinfo_url)
  86     domain_helper.raise_on(domain)
  87
  88     if not isinstance(nodeinfo_url, str):
  89         raise ValueError(f"Parameter nodeinfo_url[]='{type(nodeinfo_url)}' is not 'str'")
  90     elif nodeinfo_url == "":
  91         raise ValueError("Parameter 'nodeinfo_url' is empty")
  92
  93     translations = [
  94         "Blocked Instances".lower(),
  95         "Instàncies bloquejades".lower(),
  96         "Blocáilte Ásc".lower(),
  97         "封锁实例".lower(),
  98         "Blokované instance".lower(),
  99         "Geblokkeerde instanties".lower(),
 100         "Blockerade instanser".lower(),
 101         "Instàncias blocadas".lower(),
 102         "Istanze bloccate".lower(),
 103         "Instances bloquées".lower(),
 104         "Letiltott példányok".lower(),
 105         "Instancias bloqueadas".lower(),
 106         "Blokeatuta dauden instantziak".lower(),
 107         "차단된 인스턴스".lower(),
 108         "Peladen Yang Diblokir".lower(),
 109         "Blokerede servere".lower(),
 110         "Blokitaj nodoj".lower(),
 111         "Блокирани Инстанции".lower(),
 112         "Blockierte Instanzen".lower(),
 113         "Estetyt instanssit".lower(),
 114         "Instâncias bloqueadas".lower(),
 115         "Zablokowane instancje".lower(),
 116         "Blokované inštancie".lower(),
 117         "المثلاء المحجوبون".lower(),
 118         "Užblokuoti serveriai".lower(),
 119         "ブロックしたインスタンス".lower(),
 120         "Блокированные Инстансы".lower(),
 121         "Αποκλεισμένοι διακομιστές".lower(),
 122         "封鎖站台".lower(),
 123         "Instâncias bloqueadas".lower(),
 124     ]
 125
 126     blocklist = list()
 127
 128     try:
 129         # json endpoint for newer mastodongs
 130         logger.debug("Fetching /instances from domain='%s'", domain)
 131         response = network.fetch_response(
 132             domain,
 133             "/instances",
 134             network.web_headers,
 135             (config.get("connection_timeout"), config.get("read_timeout"))
 136         )
 137
 138         logger.debug("response.ok='%s',response.status_code=%d,response.text()=%d", response.ok, response.status_code, len(response.text))
 139         if response.ok and response.status_code < 300 and response.text != "":
 140             logger.debug("Parsing %s Bytes ...", len(response.text))
 141
 142             doc = bs4.BeautifulSoup(response.text, "html.parser")
 143             logger.debug("doc[]='%s'", type(doc))
 144
 145             headers = doc.findAll("h5")
 146             found = None
 147             logger.debug("Search in %d header(s) ...", len(headers))
 148             for header in headers:
 149                 logger.debug("header[]='%s'", type(header))
 150                 content = header.contents[0]
 151
 152                 logger.debug("content[%s]='%s'", type(content), content)
 153                 if content is None:
 154                     logger.debug("domain='%s' has returned empty header='%s' - SKIPPED!", domain, header)
 155                     continue
 156                 elif content.lower() in translations:
 157                     logger.debug("Found header with blocked instances - BREAK!")
 158                     found = header
 159                     break
 160
 161             logger.debug("found[]='%s'", type(found))
 162             if found is None:
 163                 logger.debug("domain='%s' is not blocking any instances - EXIT!", domain)
 164                 return blocklist
 165
 166             blocking = found.find_next("ul").findAll("a")
 167             logger.debug("Found %d blocked instance(s) ...", len(blocking))
 168             for tag in blocking:
 169                 logger.debug("tag[]='%s'", type(tag))
 170                 blocked = tidyup.domain(tag.contents[0])
 171                 logger.debug("blocked='%s'", blocked)
 172
 173                 if not utils.is_domain_wanted(blocked):
 174                     logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
 175                     continue
 176
 177                 logger.debug("Appending blocker='%s',blocked='%s',block_level='reject'", domain, blocked)
 178                 blocklist.append({
 179                     "blocker"    : domain,
 180                     "blocked"    : blocked,
 181                     "reason"     : None,
 182                     "block_level": "reject",
 183                 })
 184
 185     except network.exceptions as exception:
 186         logger.warning("domain='%s',exception[%s]:'%s'", domain, type(exception), str(exception))
 187         instances.set_last_error(domain, exception)
 188
 189     logger.debug("blocklist()=%d - EXIT!", len(blocklist))
 190     return blocklist