fba/helpers/domain.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 from urllib.parse import urlparse
  20
  21 import validators
  22
  23 from fba.helpers import blacklist
  24 from fba.helpers import config
  25
  26 from fba.models import instances
  27
  28 logging.basicConfig(level=logging.INFO)
  29 logger = logging.getLogger(__name__)
  30
  31 # In-function cache
  32 _cache = {
  33     # Cache for function is_in_url()
  34     "is_in_url": {},
  35
  36     # Cache for function is_wanted()
  37     "is_wanted": {},
  38
  39     # Cache for function raise_on()
  40     "raise_on": {},
  41 }
  42
  43 def raise_on(domain: str):
  44     logger.debug("domain='%s' - CALLED!", domain)
  45
  46     if not isinstance(domain, str):
  47         raise ValueError(f"Parameter domain[]='{type(domain)}' is not of type 'str'")
  48     elif domain == "":
  49         raise ValueError("Parameter 'domain' is empty")
  50     elif domain in _cache["raise_on"]:
  51         logger.debug("Returning cached raised_on='%s' - EXIT!", _cache["raise_on"][domain])
  52         return _cache["raise_on"][domain]
  53     elif domain.lower() != domain:
  54         raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
  55     elif not validators.domain(domain.split("/")[0]):
  56         raise ValueError(f"domain='{domain}' is not a valid domain")
  57     elif domain.endswith(".onion"):
  58         raise ValueError(f"domain='{domain}' is a TOR, please don't crawl them!")
  59     elif domain.endswith(".i2p") and config.get("allow_i2p_domain") == "true":
  60         raise ValueError(f"domain='{domain}' is an I2P, please don't crawl them!")
  61     elif domain.endswith(".arpa"):
  62         raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
  63     elif domain.endswith(".tld"):
  64         raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
  65
  66     _cache["raise_on"][domain] = True
  67     logger.debug("EXIT!")
  68
  69 def is_in_url(domain: str, url: str) -> bool:
  70     logger.debug("domain='%s',url='%s' - CALLED!", domain, url)
  71     raise_on(domain)
  72
  73     if blacklist.is_blacklisted(domain):
  74         raise ValueError(f"domain='{domain}' is blacklisted but function was invoked")
  75     elif not isinstance(url, str):
  76         raise ValueError(f"Parameter url[]='{type(url)}' is not of type 'str'")
  77     elif url == "":
  78         raise ValueError("Parameter 'url' is empty")
  79     elif domain + url in _cache["is_in_url"]:
  80         logger.debug("Returning cached is_in_url='%s' - EXIT!", _cache["is_in_url"][domain + url])
  81         return _cache["is_in_url"][domain + url]
  82
  83     punycode = domain.encode("idna").decode("utf-8")
  84
  85     components = urlparse(url)
  86     logger.debug("components[]='%s',punycode='%s'", type(components), punycode)
  87
  88     is_found = (punycode in [components.netloc, components.hostname])
  89
  90     # Set cache
  91     _cache["is_in_url"][domain + url] = is_found
  92
  93     logger.debug("is_found='%s' - EXIT!", is_found)
  94     return is_found
  95
  96 def is_wanted(domain: str) -> bool:
  97     logger.debug("domain='%s' - CALLED!", domain)
  98
  99     if not isinstance(domain, str):
 100         raise ValueError(f"Parameter domain[]='{type(domain)}' is not of type 'str'")
 101     elif domain == "":
 102         raise ValueError("Parameter 'domain' is empty")
 103     elif domain in _cache["is_wanted"]:
 104         logger.debug("Returning cached is_found='%s' - EXIT!", _cache["is_wanted"][domain])
 105         return _cache["is_wanted"][domain]
 106
 107     wanted = True
 108     if domain.lower() != domain:
 109         logger.debug("domain='%s' is not all-lowercase - setting False ...", domain)
 110         wanted = False
 111     elif not validators.domain(domain.split("/")[0]):
 112         logger.debug("domain='%s' is not a valid domain name - setting False ...", domain)
 113         wanted = False
 114     elif domain.endswith(".arpa"):
 115         logger.debug("domain='%s' is a domain for reversed IP addresses - setting False ...", domain)
 116         wanted = False
 117     elif domain.endswith(".onion"):
 118         logger.debug("domain='%s' is a TOR .onion domain - setting False ...", domain)
 119         wanted = False
 120     elif domain.endswith(".i2p") and config.get("allow_i2p_domain") == "true":
 121         logger.debug("domain='%s' is an I2P domain - setting False ...", domain)
 122         wanted = False
 123     elif domain.endswith(".tld"):
 124         logger.debug("domain='%s' is a fake domain - setting False ...", domain)
 125         wanted = False
 126     elif blacklist.is_blacklisted(domain):
 127         logger.debug("domain='%s' is blacklisted - setting False ...", domain)
 128         wanted = False
 129     elif domain.find("/profile/") > 0 or domain.find("/users/") > 0 or (instances.is_registered(domain.split("/")[0]) and domain.find("/c/") > 0):
 130         logger.debug("domain='%s' is a single user", domain)
 131         wanted = False
 132     elif domain.find("/tag/") > 0:
 133         logger.debug("domain='%s' is a tag", domain)
 134         wanted = False
 135
 136     # Set cache
 137     _cache["is_wanted"][domain] = wanted
 138
 139     logger.debug("wanted='%s' - EXIT!", wanted)
 140     return wanted