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