def delete_txt_record()

in lemur/plugins/lemur_acme/powerdns.py [0:0]

• 68 lines of code
• 7 McCabe index (conditional complexity)

def delete_txt_record(change_id, account_number, domain, token):
    """
    Delete the TXT record for the given domain and token

    :param change_id: tuple of domain/token
    :param account_number:
    :param domain: FQDN
    :param token: challenge to delete
    :return:
    """
    _check_conf()

    function = sys._getframe().f_code.co_name
    log_data = {
        "function": function,
        "fqdn": domain,
        "token": token,
    }

    """
    Get existing TXT records matching the domain from DNS
    The token to be deleted should already exist
    There may be other records with different tokens as well
    """
    cur_records = _get_txt_records(domain)
    found = False
    new_records = []
    for record in cur_records:
        if record.content == f"\"{token}\"":
            found = True
        else:
            new_records.append(record)

    # Since the matching token is not in DNS, there is nothing to delete
    if not found:
        log_data["message"] = "Unable to delete TXT record: Token not found in existing TXT records"
        current_app.logger.debug(log_data)
        return

    # The record to delete has been found AND there are other tokens set on the same domain
    # Since we only want to delete one token value from the RRSet, we need to use the Patch command to
    # overwrite the current RRSet with the existing records.
    elif new_records:
        try:
            _patch_txt_records(domain, account_number, new_records)
            log_data["message"] = "TXT record successfully deleted"
            current_app.logger.debug(log_data)
        except Exception as e:
            sentry.captureException()
            log_data["Exception"] = e
            log_data["message"] = "Unable to delete TXT record: patching exception"
            current_app.logger.debug(log_data)

    # The record to delete has been found AND there are no other token values set on the same domain
    # Use the Delete command to delete the whole RRSet.
    else:
        zone_name = _get_zone_name(domain, account_number)
        server_id = current_app.config.get("ACME_POWERDNS_SERVERID", "localhost")
        zone_id = zone_name + "."
        domain_id = domain + "."
        path = f"/api/v1/servers/{server_id}/zones/{zone_id}"
        payload = {
            "rrsets": [
                {
                    "name": domain_id,
                    "type": "TXT",
                    "ttl": 300,
                    "changetype": "DELETE",
                    "records": [
                        {
                            "content": f"\"{token}\"",
                            "disabled": False
                        }
                    ],
                    "comments": []
                }
            ]
        }
        function = sys._getframe().f_code.co_name
        log_data = {
            "function": function,
            "fqdn": domain,
            "token": token
        }
        try:
            _patch(path, payload)
            log_data["message"] = "TXT record successfully deleted"
            current_app.logger.debug(log_data)
        except Exception as e:
            sentry.captureException()
            log_data["Exception"] = e
            log_data["message"] = "Unable to delete TXT record"
            current_app.logger.debug(log_data)