netcache.py

#!/usr/bin/env python3
import argparse
import codecs
import datetime
import getpass
import glob
import hashlib
import os
import socket
import ssl
import sys
import time
import urllib.parse
import warnings
from ssl import CertificateError
import gettext
import subprocess

import ansicat
import offutils
from offutils import xdg, _LOCALE_DIR, get_url_redirected, edit_file, clean_url
import offblocklist

gettext.bindtextdomain('offpunk', _LOCALE_DIR)
gettext.textdomain('offpunk')
_ = gettext.gettext

def load_CRYPTOGRAPHY():
    try:
        global x509, default_backend, hashes, serialization, rsa
        from cryptography import x509
        from cryptography.hazmat.backends import default_backend
        from cryptography.hazmat.primitives import hashes, serialization
        from cryptography.hazmat.primitives.asymmetric import rsa
        _HAS_CRYPTOGRAPHY = True
    except (ModuleNotFoundError, ImportError):
        _HAS_CRYPTOGRAPHY = False
    return _HAS_CRYPTOGRAPHY

def load_HTTP():
    if offutils.CMDS["curl"]:
        return True
    return False

# For the list of supported protocols, see "PROTOCOLS" below
CRLF = "\r\n"
DEFAULT_TIMEOUT = 10
_MAX_REDIRECTS = 5

# monkey-patch Gemini support in urllib.parse
# see https://github.com/python/cpython/blob/master/Lib/urllib/parse.py
urllib.parse.uses_relative.append("gemini")
urllib.parse.uses_netloc.append("gemini")
urllib.parse.uses_relative.append("spartan")
urllib.parse.uses_netloc.append("spartan")


class UserAbortException(Exception):
    pass


def parse_mime(mime):
    options = {}
    if mime:
        if ";" in mime:
            splited = mime.split(";", maxsplit=1)
            mime = splited[0]
            if len(splited) >= 1:
                options_list = splited[1].split()
                for o in options_list:
                    spl = o.split("=", maxsplit=1)
                    if len(spl) > 0:
                        options[spl[0]] = spl[1]
    return mime, options


def normalize_url(url):
    if "://" not in url and ("./" not in url and url[0] != "/"):
        if not url.startswith("mailto:"):
            url = "gemini://" + url
    return url

def multi_line_input(prompt, url="", meta="", options={}):
    print (_("Ctrl+c to cancel. Press 'enter' on a blank line to open an external editor"))
    text = input(prompt)
    if text == "":
        text = external_editor_input(url, meta, options)
    return urllib.parse.quote(text)

def external_editor_input(url, meta, options):
    done = False
    text = ""
    initial_message = ""
    initial_message += _("|# Editing input for url: ") + str(url) + '\n'
    initial_message += _("|# The site said: ") + str(meta) + '\n'
    initial_message += _("|# (lines starting with '|#' will be deleted)") + '\n'
    initial_message += _("|# ") + '\n'
    initial_message = initial_message.encode("utf-8")
    while not done:
        to_add = initial_message + text.encode("utf-8")
        text = ""
        # calling the new function
        raw_text = edit_file(None, to_add, options)

        if not raw_text is None:
            for line in raw_text:
                if not line.startswith(b"|#"):
                    text += line.decode("utf-8")
            print(text)
            response = ""
            while response not in ("a","e","c"):
                response = input(_("(a)ccept, (e)dit again, (c)ancel? "))
                response = response.lower()
                if response == "a":
                    done = True
                elif response == "c":
                    text = ""
                    done = True
                elif response == "e":
                    break
                else:
                    pass
        else:
            # There was a problem with $EDITOR, or we can't get the file contents
            text = ""
            done = True
    return text

def cache_last_modified(url):
    if not url:
        return None
    path = get_cache_path(url)
    if path and os.path.isfile(path):
        return os.path.getmtime(path)
    else:
        return None


def is_cache_valid(url, validity=0):
    # Validity is the acceptable time for
    # a cache to be valid  (in seconds)
    # If 0, then any cache is considered as valid
    # (use validity = 1 if you want to refresh everything)
    if offutils.is_local(url):
        return True
    cache = get_cache_path(url)
    if cache:
        # If path is too long, we always return True to avoid
        # fetching it.
        if len(cache) > 259:
            print(_("We return False because path is too long"))
            return False
        if os.path.exists(cache) and not os.path.isdir(cache):
            if validity > 0:
                last_modification = cache_last_modified(url)
                now = time.time()
                age = now - last_modification
                return age < validity
            else:
                return True
        else:
            # Cache has not been build
            return False
    else:
        # There’s not even a cache!
        return False

# get_cache_path gives a local path unique to each URL
def get_cache_path(url, add_index=True, include_protocol=True, xdgfolder="cache",subfolder=""):
    # Sometimes, cache_path became a folder! (which happens for index.html/index.gmi)
    # In that case, we need to reconstruct it
    # if add_index=False, we don’t add that "index.gmi" at the ends of the cache_path
    # include protocol will include the part before :// in the path
    # xdgfolder is the XDG folder to be used
    # subfolder is the optional folder inside XDG/offpunk/
    # First, we parse the URL
    if not url:
        return None
    url = clean_url(url)
    parsed = urllib.parse.urlparse(url)
    if url[0] == "/" or url.startswith("./") or os.path.exists(url):
        scheme = "file"
    elif parsed.scheme:
        scheme = parsed.scheme
    else:
        scheme = default_protocol
    if scheme in ["file", "mailto", "list"]:
        local = True
        host = ""
        # file:// is 7 char
        if url.startswith("file://"):
            path = url[7:]
        elif scheme == "mailto":
            path = parsed.path
        elif url.startswith("list://"):
            listdir = os.path.join(xdg("data"), "lists")
            listname = url[7:].lstrip("/")
            if listname in [""]:
                name = "My Lists"
                path = listdir
            else:
                name = listname
                path = os.path.join(listdir, "%s.gmi" % listname)
        else:
            path = url
    #elif scheme in PROTOCOLS:
    else:
        local = False
        # Convert unicode hostname to punycode using idna RFC3490
        host = parsed.netloc  # .encode("idna").decode()
        # special gopher selector case
        if scheme == "gopher":
            if len(parsed.path) >= 2:
                # we remove the selector
                splitted = parsed.path.split("/")
                # the first split if empty
                if len(splitted[0]) == 0: splitted.pop(0)
                #now we see if the element is a selector on not
                if len(splitted[0]) == 1:
                    path = parsed.path[2:]
                else: 
                    path = parsed.path
            else:
                path = ""
        else:
            path = parsed.path
        if parsed.query:
            # we don’t add the query if path is too long because path above 260 char
            # are not supported and crash python.
            # Also, very long query are usually useless stuff
            if len(path + parsed.query) < 258:
                path += "/" + parsed.query
    #else:
        # print("%s is not a supported protocol"%scheme)
        # return None

    # Now, we have a partial path. Let’s make it full path.
    if local:
        cache_path = path
    elif scheme and host:
        # schemepath should not start with / but finish with /
        schemepath = ""
        if include_protocol: schemepath = scheme + "/"
        if subfolder: schemepath += subfolder + "/"
        cache_path = os.path.expanduser(xdg(xdgfolder) + schemepath + host + path)
        # There’s an OS limitation of 260 characters per path.
        # We will thus cut the path enough to add the index afterward
        cache_path = cache_path[:249]
        # this is a gross hack to give a name to
        # index files. This will break if the index is not
        # index.gmi. I don’t know how to know the real name
        # of the file. But first, we need to ensure that the domain name
        # finish by "/". Else, the cache will create a file, not a folder.
        if scheme.startswith("http"):
            index = "index.html"
        elif scheme == "finger":
            index = "index.txt"
        elif scheme == "gopher":
            index = "gophermap"
        else:
            index = "index.gmi"
        if path == "" or os.path.isdir(cache_path):
            if not cache_path.endswith("/"):
                cache_path += "/"
            if not url.endswith("/"):
                url += "/"
        if add_index and cache_path.endswith("/"):
            cache_path += index
        # sometimes, the index itself is a dir
        # like when folder/index.gmi?param has been created
        # and we try to access folder
        if add_index and os.path.isdir(cache_path):
            cache_path += "/" + index
    else:
        # URL is missing either a supported scheme or a valid host
        # print("Error: %s is not a supported url"%url)
        return None
    if len(cache_path) > 259:
        #print("Path is %s characters long which is too long. \
        #        OS only allows 260 characters.\n\n"%(len(cache_path)))
        #print(url)
        #return None
        # path length is limited to 260 characters. Let’s cut it and
        # hope that there’s no major conflict here. (that’s still better
        # than crashing, after all.
        cache_path = cache_path[:259]
    return cache_path


def write_body(url, body, mime=None):
    # body is a copy of the raw gemtext
    # Write_body() also create the cache !
    # DEFAULT GEMINI MIME
    mime, options = parse_mime(mime)
    cache_path = get_cache_path(url)
    if cache_path:
        if mime and mime.startswith("text/"):
            mode = "w"
        else:
            mode = "wb"
        cache_dir = os.path.dirname(cache_path)
        # If the subdirectory already exists as a file (not a folder)
        # We remove it (happens when accessing URL/subfolder before
        # URL/subfolder/file.gmi.
        # This causes loss of data in the cache
        # proper solution would be to save "subfolder" as "subfolder/index.gmi"
        # If the subdirectory doesn’t exist, we recursively try to find one
        # until it exists to avoid a file blocking the creation of folders
        root_dir = cache_dir
        while not os.path.exists(root_dir):
            root_dir = os.path.dirname(root_dir)
        if os.path.isfile(root_dir):
            os.remove(root_dir)
        os.makedirs(cache_dir, exist_ok=True)
        with open(cache_path, mode=mode) as f:
            f.write(body)
            f.close()
        return cache_path


def set_error(url, err):
    # If we get an error, we want to keep an existing cache
    # but we need to touch it or to create an empty one
    # to avoid hitting the error at each refresh
    cache = get_cache_path(url)
    if is_cache_valid(url):
        os.utime(cache)
    elif cache:
        cache_dir = os.path.dirname(cache)
        root_dir = cache_dir
        while not os.path.exists(root_dir):
            root_dir = os.path.dirname(root_dir)
        if os.path.isfile(root_dir):
            os.remove(root_dir)
        os.makedirs(cache_dir, exist_ok=True)
        if os.path.isdir(cache_dir):
            with open(cache, "w") as c:
                c.write(str(datetime.datetime.now()) + "\n")
                c.write(_("ERROR while caching %s\n\n") % url)
                c.write("*****\n\n")
                c.write(str(type(err)) + " = " + str(err))
                # cache.write("\n" + str(err.with_traceback(None)))
                c.write("\n*****\n\n")
                c.write(_("If you believe this error was temporary, type " "reload" ".\n"))
                c.write(_("The resource will be tentatively fetched during next sync.\n"))
                c.close()
    return cache


def get_cookiejar(url, create=False):
    if load_HTTP():
        parsed = urllib.parse.urlparse(url)
        basedir = os.path.join(xdg("data") + "cookies/" )
        cookie_path = basedir + parsed.netloc + ".txt"
        if not os.path.exists(cookie_path):
            if create:
                os.makedirs(basedir, exist_ok=True)
                with open(cookie_path, 'w') as f:
                    f.write("# Netscape HTTP Cookie File\n\n")
            else:
                return None
        import http.cookiejar
        jar = http.cookiejar.MozillaCookieJar(cookie_path)
        jar.load()
        return jar
    else:
        return None

#23: CURLE_WRITE_ERROR
CURL_WRITE_ERROR = 23
# 60: Peer certificate cannot be authenticated with known CA certificates.
CURL_BAD_SSL = 60
# 63: Maximum file size exceeded
CURL_MAX_FILE_SIZE_EXCEEDED = 63

class CurlError(Exception):
    pass

def _fetch_curl(url, verify=True, headers={}, timeout=DEFAULT_TIMEOUT, cookies=None, max_size=None, extra_args=[]):
    """
    return cache path
    can raise CurlError(exitcode, error message)
    """
    if offutils.CMDS["curl"] is None:
        return None

    import http.cookiejar
    use_cookie = cookies is not None and isinstance(cookies, http.cookiejar.MozillaCookieJar)

    def too_large_error(url, length, max_size):
        err = _("Size of %s is %s Mo\n") % (url, length)
        err += _("Offpunk only download automatically content under %s Mo\n") % (
            max_size / 1000000
        )
        err += _("To retrieve this content anyway, type 'reload'.")
        return set_error(url, err)

    # -b : use cookies from file
    # -c : save cookies to file
    # -k : accept bad server certificate
    # -o : write output to file, single URL per file
    # -s : don't show anything
    # -w : write format string after a request is finished or failed
    # -A : set user agent
    # -D : dump response header to file
    # -H : set header
    # -L : redirect
    # -S : print error to stderr
    # -Z : use parallel request
    # -z : download only if remote file is newer than the local one
    # --connect-timeout
    # --create-dirs      creates the necessary local directory hierarchy as needed
    # --max-filesize
    cmd = [offutils.CMDS["curl"], "-sSL", "--compressed", "--create-dirs"]
    cmd += ["-A", "Offpunk/Netcache - https://offpunk.net"]
    for key in headers:
        cmd += ["-H", f"{key}: {headers[key]}"]
    if not verify:
        cmd += ["-k"]
    if timeout:
        cmd += ["--connect-timeout", str(timeout)]
    if max_size:
        cmd += ["--max-filesize", str(max_size)]
    if use_cookie:
        # save all cookies for curl to read
        cookies.save()
        cmd += ["-c", cookies.filename]
        cmd += ["-b", cookies.filename]
    # curl only write its output as-is, it DOES NOT re-encode output to utf-8 like python-requests.
    # Response with non-unicode charset may crash with UnicodeDecodeError when python read the output file.
    cache = get_cache_path(url)
    cmd += [url, "-o", cache]
    # save the Last-Modified time as the file modification time
    cmd += ["--remote-time"]
    # if we already have it, make a conditional request
    if os.path.isfile(cache):
        cmd += ["-z", cache]
    cmd += extra_args
    try:
        subprocess.run(cmd, capture_output=True, check=True)
    except Exception as err:
        if err.returncode == CURL_MAX_FILE_SIZE_EXCEEDED:
            return too_large_error(url, length, max_size)
        elif err.returncode == CURL_WRITE_ERROR:
            #There’s a write error so we don’t have any cache, we return none
            return None
        else:
            raise CurlError(err.returncode, err.stderr.decode().strip())
    return cache, url

def _fetch_http(
    url,
    max_size=None,
    timeout=DEFAULT_TIMEOUT,
    accept_bad_ssl_certificates=False,
    force_large_download=False,
    cookiejar=None,
    **kwargs,
):
    if not load_HTTP():
        print(_("HTTP requires curl"))
        return None
    if not cookiejar:
        cookiejar = get_cookiejar(url)
    verify = not accept_bad_ssl_certificates
    if force_large_download:
        max_size = None
    return _fetch_curl(url=url, verify=verify, timeout=timeout,
                cookies=cookiejar, max_size=max_size)

def _fetch_gopher(url, timeout=DEFAULT_TIMEOUT, interactive=True, **kwargs):
    parsed = urllib.parse.urlparse(url)
    if len(parsed.path) >= 2:
        itemtype = parsed.path[1]
    else:
        itemtype = "1"

    if itemtype == "7":
        if interactive:
            user_input = input("> ")
            url = url+'?'+user_input
        else:
            return None, url

    return _fetch_curl(url=url, timeout=timeout)

def _fetch_finger(url, timeout=DEFAULT_TIMEOUT, **kwargs):
    parsed = urllib.parse.urlparse(url)
    host = parsed.hostname
    port = parsed.port or PROTOCOLS["finger"]["port"]
    query = parsed.path.lstrip("/") + "\r\n"
    with socket.create_connection((host, port)) as sock:
        sock.settimeout(timeout)
        sock.send(query.encode())
        response = sock.makefile("rb").read().decode("UTF-8")
        cache = write_body(response, "text/plain")
    return cache, url


# Originally copied from reference spartan client by Michael Lazar
def _fetch_spartan(url, **kwargs):
    cache = None
    url_parts = urllib.parse.urlparse(url)
    host = url_parts.hostname
    port = url_parts.port or PROTOCOLS["spartan"]["port"]
    path = url_parts.path or "/"
    query = url_parts.query
    redirect_url = None
    with socket.create_connection((host, port)) as sock:
        if query:
            data = urllib.parse.unquote_to_bytes(query)
        else:
            data = b""
        encoded_host = host.encode("idna")
        ascii_path = urllib.parse.unquote_to_bytes(path)
        encoded_path = urllib.parse.quote_from_bytes(ascii_path).encode("ascii")
        sock.send(b"%s %s %d\r\n" % (encoded_host, encoded_path, len(data)))
        fp = sock.makefile("rb")
        response = fp.readline(4096).decode("ascii").strip("\r\n")
        parts = response.split(" ", maxsplit=1)
        code, meta = int(parts[0]), parts[1]
        if code == 2:
            body = fp.read()
            if meta.startswith("text"):
                body = body.decode("UTF-8")
            cache = write_body(url, body, meta)
        elif code == 3:
            redirect_url = url_parts._replace(path=meta).geturl()
        else:
            return set_error(url, "Spartan code %s: Error %s" % (code, meta))
    if redirect_url:
        cache = _fetch_spartan(redirect_url)
    return cache


def _validate_cert(address, host, cert, accept_bad_ssl=False, automatic_choice=None):
    """
    Validate a TLS certificate in TOFU mode.

    If the cryptography module is installed:
     - Check the certificate Common Name or SAN matches `host`
     - Check the certificate's not valid before date is in the past
     - Check the certificate's not valid after date is in the future

    Whether the cryptography module is installed or not, check the
    certificate's fingerprint against the TOFU database to see if we've
    previously encountered a different certificate for this IP address and
    hostname.
    """
    now = datetime.datetime.now(datetime.timezone.utc)
    if load_CRYPTOGRAPHY():
        # Using the cryptography module we can get detailed access
        # to the properties of even self-signed certs, unlike in
        # the standard ssl library...
        c = x509.load_der_x509_certificate(cert, default_backend())
        # Check certificate validity dates
        if accept_bad_ssl:
            if c.not_valid_before >= now:
                raise CertificateError(
                    _("Certificate not valid until: {}!").format(c.not_valid_before)
                )
            elif c.not_valid_after <= now:
                raise CertificateError(
                    _("Certificate expired as of: {})!").format(c.not_valid_after)
                )

        # Check certificate hostnames
        names = []
        common_name = c.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME)
        if common_name:
            names.append(common_name[0].value)
        try:
            names.extend(
                [
                    alt.value
                    for alt in c.extensions.get_extension_for_oid(
                        x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME
                    ).value
                ]
            )
        except x509.ExtensionNotFound:
            pass
        names = set(names)
        for name in names:
            try:
                ssl._dnsname_match(str(name), host)
                break
            except CertificateError:
                continue
        else:
            # If we didn't break out, none of the names were valid
            raise CertificateError(
                _("Hostname does not match certificate common name or any alternative names.")
            )

    sha = hashlib.sha256()
    sha.update(cert)
    fingerprint = sha.hexdigest()

    # The directory of this host and IP-address, e.g.
    # ~/.local/share/offpunk/certs/srht.site/46.23.81.157/
    certdir = os.path.join(xdg("data"), "certs")
    hostdir = os.path.join(certdir, host)
    sitedir = os.path.join(hostdir, address)
    # 1. We check through cached certificates do extract the
    # most_frequent_cert  and to see if one is matching the current one.
    # 2. If we have no match but one valid most_frequent_cert, we do the
    # "throws warning" code.
    # 3. If no certificate directory or no valid cached certificates, we do
    # the "First-Use" routine.
    most_frequent_cert = None
    matching_fingerprint = False
    # 1. Have we been here before? (the directory exists)
    if os.path.isdir(sitedir):
        max_count = 0
        files = os.listdir(sitedir)
        count = 0
        certcache = os.path.join(xdg("config"), "cert_cache")

        for cached_fingerprint in files:
            filepath = os.path.join(sitedir, cached_fingerprint)
            certpath = os.path.join(certcache, cached_fingerprint + ".crt")
            with open(filepath, "r") as f:
                count = int(f.read())
            if os.path.exists(certpath):
                if count > max_count:
                    max_count = count
                    most_frequent_cert = cached_fingerprint
                if fingerprint == cached_fingerprint:
                    # Matched!
                    # Increase the counter for this certificate (this also updates
                    # the modification time of the file)
                    with open(filepath, "w") as f:
                        f.write(str(count + 1))
                    matching_fingerprint = True
                    break
    # 2. Do we have some certificates but none of them is matching the current one?
    if most_frequent_cert and not matching_fingerprint:
        with open(os.path.join(certcache, most_frequent_cert + ".crt"), "rb") as fp:
            previous_cert = fp.read()
        if load_CRYPTOGRAPHY():
            # Load the most frequently seen certificate to see if it has
            # expired
            previous_cert = x509.load_der_x509_certificate(previous_cert, default_backend())
            previous_ttl = previous_cert.not_valid_after_utc - now
            print(previous_ttl)

        print("****************************************")
        print(_("[SECURITY WARNING] Unrecognised certificate!"))
        print(
            _("The certificate presented for {} ({}) has never been seen before.").format(
                host, address
            )
        )
        print(_("This MIGHT be a Man-in-the-Middle attack."))
        print(
            _("A different certificate has previously been seen {} times.").format(
                max_count
            )
        )
        if load_CRYPTOGRAPHY():
            if previous_ttl < datetime.timedelta():
                print(_("That certificate has expired, which reduces suspicion somewhat."))
            else:
                print(_("That certificate is still valid for: {}").format(previous_ttl))
        print("****************************************")
        print(_("Attempt to verify the new certificate fingerprint out-of-band:"))
        print(fingerprint)
        if automatic_choice:
            choice = automatic_choice
        else:
            #TRANSLATORS: keep "Y/N" because the answer has to be one of those
            choice = input(_("Accept this new certificate? Y/N ")).strip().lower()
        if choice in ("y", "yes"):
            with open(os.path.join(sitedir, fingerprint), "w") as fp:
                fp.write("1")
            with open(os.path.join(certcache, fingerprint + ".crt"), "wb") as fp:
                fp.write(cert)
        else:
            raise Exception(_("TOFU Failure!"))

    # 3. If no directory or no cert found in it, we cache it
    if not most_frequent_cert:
        if not os.path.exists(certdir):  # XDG_DATA/offpunk/certs
            os.makedirs(certdir)
        if not os.path.exists(hostdir):  # XDG_DATA/offpunk/certs/site.net
            os.makedirs(hostdir)
        if not os.path.exists(
            sitedir
        ):  # XDG_DATA/offpunk/certs/site.net/123.123.123.123
            os.makedirs(sitedir)

        with open(os.path.join(sitedir, fingerprint), "w") as fp:
            fp.write("1")
        certcache = os.path.join(xdg("config"), "cert_cache")
        if not os.path.exists(certcache):
            os.makedirs(certcache)
        with open(os.path.join(certcache, fingerprint + ".crt"), "wb") as fp:
            fp.write(cert)


def _get_client_certkey(site_id: str, host: str):
    # returns {cert: str, key: str}
    certdir = os.path.join(xdg("data"), "certs", host)
    certf = os.path.join(certdir, "%s.cert" % site_id)
    keyf = os.path.join(certdir, "%s.key" % site_id)
    if not os.path.exists(certf) or not os.path.exists(keyf):
        if host != "":
            split = host.split(".")
            # if len(split) > 2:  # Why not allow a global identity? Maybe I want
            # to login to all sites with the same
            # certificate.
            return _get_client_certkey(site_id, ".".join(split[1:]))
        return None
    certkey = dict(cert=certf, key=keyf)
    return certkey


def _get_site_ids(url: str):
    newurl = normalize_url(url)
    u = urllib.parse.urlparse(newurl)
    if u.scheme == "gemini" and u.username is None:
        certdir = os.path.join(xdg("data"), "certs")
        netloc_parts = u.netloc.split(".")
        site_ids = []

        for i in range(len(netloc_parts), 0, -1):
            lasti = ".".join(netloc_parts[-i:])
            direc = os.path.join(certdir, lasti)

            for certfile in glob.glob(os.path.join(direc, "*.cert")):
                site_id = certfile.split("/")[-1].split(".")[-2]
                site_ids.append(site_id)
        return site_ids
    else:
        return []


def create_certificate(name: str, days: int, hostname: str):
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    sitecertdir = os.path.join(xdg("data"), "certs", hostname)
    keyfile = os.path.join(sitecertdir, name + ".key")
    # create the directory of it doesn't exist
    os.makedirs(sitecertdir, exist_ok=True)
    with open(keyfile, "wb") as f:
        f.write(
            key.private_bytes(
                encoding=serialization.Encoding.PEM,
                format=serialization.PrivateFormat.TraditionalOpenSSL,
                encryption_algorithm=serialization.NoEncryption(),
            )
        )
    xname = x509.Name(
        [
            x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, name),
        ]
    )
    # generate the cert, valid a week ago (timekeeping is hard, let's give it a
    # little margin). issuer and subject are your name
    cert = (
        x509.CertificateBuilder()
        .subject_name(xname)
        .issuer_name(xname)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(datetime.datetime.utcnow() - datetime.timedelta(days=7))
        .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=days))
        .sign(key, hashes.SHA256())
    )
    certfile = os.path.join(sitecertdir, name + ".cert")
    with open(certfile, "wb") as f:
        f.write(cert.public_bytes(serialization.Encoding.PEM))

def ask_certs(url: str):
    certs = get_certs(url)
    if len(certs) == 0:
        print(_("There are no certificates available for this site."))
        #TRANSLATORS: keep the "y/n"
        create_cert = input(_("Do you want to create one? (y/n) "))
        if create_cert == "y":
            name = input(_("Name for this certificate: "))
            days = input(_("Validity in days: "))
            if name != "" and days.isdigit():
                site = urllib.parse.urlparse(url)
                create_certificate(name, int(days), site.hostname)
                new_url = "gemini://" + name +"@"+ url.split("://")[1]
                return(new_url)
            else:
                print(_("The name or validity you typed are invalid"))
                return(url)
        else:
            return(url)
    if len(certs) == 1:
        print(_("The one available certificate for this site is:"))
    elif len(certs) > 1:
        print(
                _("The {} available certificates for this site are:").format(len(certs))
        )
    if len(certs) > 0:
        counter = 0
        stri = ""
        for cert in certs:
            stri += "[%s] %s \n" % (counter + 1, cert)
            counter += 1
        stri += "\n"
        stri += _("which certificate do you want to use? > ")
        ans = input(stri)
        if ans.isdigit() and 0 < int(ans) <= len(certs):
            identity = certs[int(ans) -1]
        else:
            identity = None
        if identity:
            new_url = "gemini://" + identity +"@"+ url.split("://")[1]
            return(new_url)
    return(url) #return the same url, no "cert" attached


def get_certs(url: str):
    u = urllib.parse.urlparse(normalize_url(url))
    if u.scheme == "gemini":
        certdir = os.path.join(xdg("data"), "certs")
        netloc_parts = u.netloc.split(".")
        site_ids = []
        if "@" in netloc_parts[0]:
            netloc_parts[0] = netloc_parts[0].split("@")[1]
        # certdir does not contemplate ports, so we should take it out here if present
        if ":" in netloc_parts[-1]:
            netloc_parts[-1] = netloc_parts[-1].split(":")[0]
        for i in range(len(netloc_parts), 0, -1):
            lasti = ".".join(netloc_parts[-i:])
            direc = os.path.join(certdir, lasti)
            for certfile in glob.glob(os.path.join(direc, "*.cert")):
                site_id = certfile.split("/")[-1].split(".")[-2]
                site_ids.append(site_id)
        return site_ids
    else:
        return []


def _fetch_gemini(
    url,
    timeout=DEFAULT_TIMEOUT,
    interactive=True,
    accept_bad_ssl_certificates=False,
    **kwargs,
):
    cache = None
    newurl = url
    url_parts = urllib.parse.urlparse(url)
    host = url_parts.hostname
    site_id = url_parts.username
    port = url_parts.port or PROTOCOLS["gemini"]["port"]
    path = url_parts.path or "/"
    query = url_parts.query
    # In AV-98, this was the _send_request method
    # Send a selector to a given host and port.
    # Returns the resolved address and binary file with the reply.
    host = host.encode("idna").decode()
    # Do DNS resolution
    # DNS lookup - will get IPv4 and IPv6 records if IPv6 is enabled
    if ":" in host:
        # This is likely a literal IPv6 address, so we can *only* ask for
        # IPv6 addresses or getaddrinfo will complain
        family_mask = socket.AF_INET6
    elif socket.has_ipv6:
        # Accept either IPv4 or IPv6 addresses
        family_mask = 0
    else:
        # IPv4 only
        family_mask = socket.AF_INET
    addresses = socket.getaddrinfo(
        host, port, family=family_mask, type=socket.SOCK_STREAM
    )
    # Sort addresses so IPv6 ones come first
    addresses.sort(key=lambda add: add[0] == socket.AF_INET6, reverse=True)
    # Continuation of send_request
    # Prepare TLS context
    protocol = (
        ssl.PROTOCOL_TLS_CLIENT if sys.version_info.minor >= 6 else ssl.PROTOCOL_TLSv1_2
    )
    context = ssl.SSLContext(protocol)
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE

    # When using an identity, use the certificate and key
    if site_id:
        certkey = _get_client_certkey(site_id, host)
        if certkey:
            context.load_cert_chain(certkey["cert"], certkey["key"])
        else:
            print(_("This identity doesn't exist for this site (or is disabled)."))
    # Impose minimum TLS version
    # In 3.7 and above, this is easy...
    if sys.version_info.minor >= 7:
        context.minimum_version = ssl.TLSVersion.TLSv1_2
    # Otherwise, it seems very hard...
    # The below is less strict than it ought to be, but trying to disable
    # TLS v1.1 here using ssl.OP_NO_TLSv1_1 produces unexpected failures
    # with recent versions of OpenSSL.  What a mess...
    else:
        context.options |= ssl.OP_NO_SSLv3
        context.options |= ssl.OP_NO_SSLv2
    # Try to enforce sensible ciphers
    try:
        context.set_ciphers(
            "AESGCM+ECDHE:AESGCM+DHE:CHACHA20+ECDHE:CHACHA20+DHE:!DSS:!SHA1:!MD5:@STRENGTH"
        )
    except ssl.SSLError:
        # Rely on the server to only support sensible things, I guess...
        pass
        # Connect to remote host by any address possible
    err = None
    for address in addresses:
        try:
            s = socket.socket(address[0], address[1])
            s.settimeout(timeout)
            s = context.wrap_socket(s, server_hostname=host)
            s.connect(address[4])
            break
        except OSError as e:
            err = e
    else:
        # If we couldn't connect to *any* of the addresses, just
        # bubble up the exception from the last attempt and deny
        # knowledge of earlier failures.
        raise err

    # Do TOFU
    cert = s.getpeercert(binary_form=True)
    # Remember that we showed the current cert to this domain...
    # TODO : accept badssl and automatic choice
    _validate_cert(address[4][0], host, cert, automatic_choice="y")
    # Send request and wrap response in a file descriptor
    url = urllib.parse.urlparse(url)
    new_host = host
    # Handle IPV6 hostname
    if ":" in new_host:
        new_host = "[" + new_host + "]"
    if port != PROTOCOLS["gemini"]["port"]:
        new_host += ":" + str(port)
    url_no_username = urllib.parse.urlunparse(url._replace(netloc=new_host))

    if site_id:
        url = urllib.parse.urlunparse(url._replace(netloc=site_id + "@" + new_host))
    else:
        url = url_no_username

    s.sendall((url_no_username + CRLF).encode("UTF-8"))
    f = s.makefile(mode="rb")
    ## end of send_request in AV98
    # Spec dictates <META> should not exceed 1024 bytes,
    # so maximum valid header length is 1027 bytes.
    header = f.readline(1027)
    header = urllib.parse.unquote(header.decode("UTF-8"))
    if not header or header[-1] != "\n":
        raise RuntimeError(_("Received invalid header from server!"))
    header = header.strip()
    # Validate header
    status, meta = header.split(maxsplit=1)
    if len(meta) > 1024 or len(status) != 2 or not status.isnumeric():
        f.close()
        raise RuntimeError(_("Received invalid header from server!"))
    # Update redirect loop/maze escaping state
    if not status.startswith("3"):
        previous_redirectors = set()
    # TODO FIXME
    else:
        # we set a previous_redirectors anyway because refactoring in progress
        previous_redirectors = set()
    # Handle non-SUCCESS headers, which don't have a response body
    # Inputs
    if status.startswith("1"):
        if interactive:
            print(meta)
            if status == "11":
                user_input = getpass.getpass("> ")
            else:
                user_input = multi_line_input("> ", newurl, meta, kwargs)
            newurl = url.split("?")[0]
            return _fetch_gemini(newurl + "?" + user_input)
        else:
            return None, None
    # Redirects
    elif status.startswith("3"):
        newurl = urllib.parse.urljoin(url, meta)
        if newurl == url:
            raise RuntimeError(_("URL redirects to itself!"))
        elif newurl in previous_redirectors:
            raise RuntimeError(_("Caught in redirect loop!"))
        elif len(previous_redirectors) == _MAX_REDIRECTS:
            raise RuntimeError(
                _("Refusing to follow more than %d consecutive redirects!")
                % _MAX_REDIRECTS
            )
        # TODO: redirections handling should be refactored
        #        elif "interactive" in options and not options["interactive"]:
        #            follow = self.automatic_choice
        #        # Never follow cross-domain redirects without asking
        #        elif new_gi.host.encode("idna") != gi.host.encode("idna"):
        #            follow = input("Follow cross-domain redirect to %s? (y/n) " % new_gi.url)
        #        # Never follow cross-protocol redirects without asking
        #        elif new_gi.scheme != gi.scheme:
        #            follow = input("Follow cross-protocol redirect to %s? (y/n) " % new_gi.url)
        #        # Don't follow *any* redirect without asking if auto-follow is off
        #        elif not self.options["auto_follow_redirects"]:
        #            follow = input("Follow redirect to %s? (y/n) " % new_gi.url)
        #        # Otherwise, follow away
        else:
            follow = "yes"
        if follow.strip().lower() not in ("y", "yes"):
            raise UserAbortException()
        previous_redirectors.add(url)
        #        if status == "31":
        #            # Permanent redirect
        #            self.permanent_redirects[gi.url] = new_gi.url
        return _fetch_gemini(newurl, interactive=interactive)
    # Errors
    elif status.startswith("4") or status.startswith("5"):
        raise RuntimeError(meta)
    # Client cert
    elif status.startswith("6"):
        if interactive:
            print(_("You need to provide a client-certificate to access this page."))
            url_with_identity = ask_certs(url)
            if (url_with_identity != url):
                return fetch(url_with_identity)
        error = _("You need to provide a client-certificate to access this page.\r\nType \"certs\" to create or re-use one")
        raise RuntimeError(error)
    # Invalid status
    elif not status.startswith("2"):
        raise RuntimeError(_("Server returned undefined status code %s!") % status)
    # If we're here, this must be a success and there's a response body
    assert status.startswith("2")
    mime = meta
    # Read the response body over the network
    fbody = f.read()
    # DEFAULT GEMINI MIME
    if mime == "":
        mime = "text/gemini; charset=utf-8"
    shortmime, mime_options = parse_mime(mime)
    if "charset" in mime_options:
        try:
            codecs.lookup(mime_options["charset"])
        except LookupError:
            # raise RuntimeError("Header declared unknown encoding %s" % mime_options)
            # If the encoding is wrong, there’s a high probably it’s UTF-8 with a bad header
            mime_options["charset"] = "UTF-8"
    if shortmime.startswith("text/"):
        # Get the charset and default to UTF-8 in none
        encoding = mime_options.get("charset", "UTF-8")
        try:
            body = fbody.decode(encoding, "replace")
        except UnicodeError:
            raise RuntimeError(
                _("Could not decode response body using %s\
                                encoding declared in header!")
                % encoding
            )
    else:
        body = fbody
    cache = write_body(url, body, mime)
    return cache, url

# This is the list of supported protocols and their method
PROTOCOLS = {
    "gemini": { "port": 1965, "fetch": _fetch_gemini},
    "gopher": { "port": 70, "fetch": _fetch_gopher},
    "finger": { "port": 79, "fetch": _fetch_finger},
    "http": { "port": 80, "fetch": _fetch_http},
    "https": { "port": 443, "fetch": _fetch_http},
    "spartan": { "port": 300, "fetch": _fetch_spartan},
}

default_protocol = "gemini"

#fetch returns two things:
#cachepath: the path to the cached resource
#newurl: the real URL of that cached resource
def fetch(
    url,
    offline=False,
    download_image_first=True,
    images_mode="readable",
    validity=0,
    cookiejar=None,
    redirects={},
    #blocked is empty by default to allow blocking rules having been removed
    blocked={},
    **kwargs,
):
    url = normalize_url(url)
    url = clean_url(url)
    newurl = url
    path = None
    print_error = "print_error" in kwargs.keys() and kwargs["print_error"]
    #Step 0: we apply redirect and/or block
    #Let’s add the blocked list into the redirects
    #This will not overwrite existing rule for that domain
    for b in blocked:
        if b not in redirects.keys(): redirects[b] = "blocked"
    redirection, key = get_url_redirected(url,redirects,returnkey=True)
    if redirection and redirection.lower() == "blocked":
        text = ""
        text += _("Blocked URL: ")+url + "\n"
        text += _("This website has been blocked with the following rule:\n")
        text += key + "\n"
        text += _("Use the following redirect command to unblock it:\n")
        text += "redirect %s NONE" %key
        if print_error:
            print(text)
        cache = set_error(newurl, text)
        return cache, newurl
    elif redirection and redirection.lower() != "whitelisted":
        parsed = urllib.parse.urlparse(url)
        parsed = parsed._replace(netloc=redirection)
        url = urllib.parse.urlunparse(parsed)
        newurl = url
    # First, we look if we have a valid cache, even if offline
    # If we are offline, any cache is better than nothing
    if is_cache_valid(url, validity=validity) or (
        offline and is_cache_valid(url, validity=0)
    ):
        path = get_cache_path(url)
        # if the cache is a folder, we should add a "/" at the end of the URL
        if not url.endswith("/") and os.path.isdir(
            get_cache_path(url, add_index=False)
        ):
            newurl = url + "/"
    elif offline and is_cache_valid(url, validity=0):
        path = get_cache_path(url)
    elif "://" in url and not offline:
        try:
            scheme = url.split("://")[0]
            if scheme not in PROTOCOLS:
                if print_error:
                    print(_("%s is not a supported protocol") % scheme)
                path = None
            elif scheme in PROTOCOLS:
                path, newurl = PROTOCOLS[scheme]["fetch"](newurl, **kwargs)
            else:
                print("scheme %s not implemented yet" % scheme)
        except UserAbortException:
            return None, newurl
        except Exception as err:
            cache = set_error(newurl, err)
            # Print an error message
            # we fail silently when sync_only
            if isinstance(err, socket.gaierror):
                if print_error:
                    print(_("ERROR: DNS error!"))
            elif isinstance(err, ConnectionRefusedError):
                if print_error:
                    print(_("ERROR1: Connection refused!"))
            elif isinstance(err, ConnectionResetError):
                if print_error:
                    print(_("ERROR2: Connection reset!"))
            elif isinstance(err, (TimeoutError, socket.timeout)):
                if print_error:
                    print(_("""ERROR3: Connection timed out!
    Slow internet connection?  Use 'set timeout' to be more patient."""))
            elif isinstance(err, FileExistsError):
                if print_error:
                    print(_("""ERROR5: Trying to create a directory which already exists
                        in the cache : """))
                print(err)
            elif load_HTTP() and isinstance(err, CurlError) and err.args[0] == CURL_BAD_SSL:
                if print_error:
                    print(_("""ERROR6: Bad SSL certificate:\n"""))
                    print(
                        _("""\n If you know what you are doing, you can try to accept bad certificates with the following command:\n""")
                    )
                    print("""set accept_bad_ssl_certificates True""")
            elif load_HTTP() and isinstance(err, CurlError):
                if print_error:
                    print(_("""ERROR7: curl exit with %s:""") % err.args[0])
                    print(err.args[1])
            else:
                if print_error:
                    import traceback

                    print(_("ERROR4: ") + str(type(err)) + " : " + str(err))
                    # print("\n" + str(err.with_traceback(None)))
                    print(traceback.format_exc())
            return cache, newurl
        # We download images contained in the document (from full mode)
        if not offline and download_image_first and images_mode:
            renderer = ansicat.renderer_from_file(path, newurl,redirectlist=redirects,**kwargs)
            if renderer:
                for image in renderer.get_images(mode=images_mode):
                    # Image should exist, should be an url (not a data image)
                    # and should not be already cached
                    if (
                        image
                        and not image.startswith("data:image/")
                        and not is_cache_valid(image)
                    ):
                        width = offutils.term_width() - 1
                        toprint = _("Downloading %s") % image
                        toprint = toprint[:width]
                        toprint += " " * (width - len(toprint))
                        print(toprint, end="\r")
                        # d_i_f and images_mode are False/None to avoid recursive downloading
                        # if that ever happen
                        fetch(
                            image,
                            offline=offline,
                            download_image_first=False,
                            images_mode=None,
                            validity=0,
                            cookiejar=cookiejar,
                            redirects=redirects,
                            **kwargs,
                        )
    if download_image_first and cookiejar is not None:
        cookiejar.save()
    return path, newurl


def main():
    descri = _("Netcache is a command-line tool to retrieve, cache and access networked content.\n\
            By default, netcache will returns a cached version of a given URL, downloading it \
            only if a cache version doesn't exist. A validity duration, in seconds, can also \
            be given so netcache downloads the content only if the existing cache is older than the validity.")
    # Parse arguments
    parser = argparse.ArgumentParser(prog="netcache", description=descri)
    parser.add_argument(
        "--path",
        action="store_true",
        help=_("return path to the cache instead of the content of the cache"),
    )
    parser.add_argument(
        "--ids",
        action="store_true",
        help=_("return a list of id's for the gemini-site instead of the content of the cache"),
    )
    parser.add_argument(
        "--offline",
        action="store_true",
        help=_("Do not attempt to download, return cached version or error"),
    )
    parser.add_argument(
        "--max-size",
        type=int,
        help=_("Cancel download of items above that size (value in Mb)."),
    )
    parser.add_argument(
        "--timeout",
        type=int,
        help=_("Time to wait before cancelling connection (in second)."),
    )
    parser.add_argument(
        "--cache-validity",
        type=int,
        default=0,
        help=_("maximum age, in second, of the cached version before \
                                redownloading a new version"),
    )
    # No argument: write help
    parser.add_argument(
        "url",
        metavar="URL",
        nargs="*",
        help=_("download URL and returns the content or the path to a cached version"),
    )
    # --validity : returns the date of the cached version, Null if no version
    # --force-download : download and replace cache, even if valid
    args = parser.parse_args()
    param = {}

    for u in args.url:
        if args.offline:
            path = get_cache_path(u)
        elif args.ids:
            ids = _get_site_ids(u)
        else:
            path, url = fetch(
                u,
                max_size=args.max_size,
                timeout=args.timeout,
                validity=args.cache_validity,
                redirects=offblocklist.redirects,
                blocked=offblocklist.blocked,
            )
        if args.path:
            print(path)
        elif args.ids:
            print(ids)
        else:
            with open(path, "r") as f:
                print(f.read())
                f.close()


if __name__ == "__main__":
    main()