org_fedora_oscap.data_fetch._curl_fetch()

Complexity

Conditions

18

Size

Total Lines	86
Code Lines	50

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

"""
def _curl_fetch(url, out_file, ca_certs_path=None):
    """
    Function that fetches data and writes it out to the given file path. If a
    path to the file with CA certificates is given and the url starts with
    'https', the server certificate is validated.

    :param url: url of the data that has to start with 'http://' or "https://"
    :type url: str
    :param out_file: path to the output file
    :type out_file: str
    :param ca_certs_path: path to the file with CA certificates for server
                     certificate validation
    :type ca_certs_path: str
    :raise WrongRequestError: if a wrong combination of arguments is passed
                              (ca_certs_path file path given and url starting with
                              http://) or arguments don't have required format
    :raise CertificateValidationError: if server certificate validation fails
    :raise FetchError: if data fetching fails (usually due to I/O errors)

    """

    if url.startswith("ftp"):
        match = FTP_URL_RE.match(url)
        if not match:
            msg = "Wrong url not matching '%s'" % FTP_URL_RE_STR
            raise WrongRequestError(msg)
        else:
            protocol, path = match.groups()
            if '@' not in path:
                # no user:pass given -> use anonymous login to the FTP server
                url = protocol + "://anonymous:@" + path
    elif url.startswith("file"):
        match = FILE_URL_RE.match(url)
        if not match:
            msg = "Wrong url not matching '%s'" % FILE_URL_RE_STR
            raise WrongRequestError(msg)
    else:
        match = HTTP_URL_RE.match(url)
        if not match:
            msg = "Wrong url not matching '%s'" % HTTP_URL_RE_STR
            raise WrongRequestError(msg)

    # the first group contains the protocol, the second one the rest
    protocol = match.groups()[0]

    if not out_file:
        raise WrongRequestError("out_file cannot be an empty string")

    if ca_certs_path and protocol != "https":
        msg = "Cannot verify server certificate when using plain HTTP"
        raise WrongRequestError(msg)

    curl = pycurl.Curl()
    curl.setopt(pycurl.URL, url)

    if ca_certs_path and protocol == "https":
        # the strictest verification
        curl.setopt(pycurl.SSL_VERIFYHOST, 2)
        curl.setopt(pycurl.SSL_VERIFYPEER, 1)
        curl.setopt(pycurl.CAINFO, ca_certs_path)

    # may be turned off by flags (specified on command line, take precedence)
    if not conf.payload.verify_ssl:
        log.warning("Disabling SSL verification due to the noverifyssl flag")
        curl.setopt(pycurl.SSL_VERIFYHOST, 0)
        curl.setopt(pycurl.SSL_VERIFYPEER, 0)

    try:
        with open(out_file, "wb") as fobj:
            curl.setopt(pycurl.WRITEDATA, fobj)
            curl.perform()
    except pycurl.error as err:
        # first arg is the error code
        if err.args[0] == pycurl.E_SSL_CACERT:
            msg = "Failed to connect to server and validate its "\
                  "certificate: %s" % err
            raise CertificateValidationError(msg)
        else:
            msg = "Failed to fetch data: %s" % err
            raise FetchError(msg)

    if protocol in ("http", "https"):
        return_code = curl.getinfo(pycurl.HTTP_CODE)
        if 400 <= return_code < 600:
            msg = _(f"Failed to fetch data - the request returned HTTP error code {return_code}")
            raise FetchError(msg)