Dns-01 challenge has been circulating for a long time, but I still haven't been able to achieve it

Client dev
1

It’s been going on for a long time, and I still can’t finish it, in which I begin to doubt the steps ahead. It may be that I don’t understand some parts of the document.

I’m not sure about authorization, so let me confirm that the key authorization string is computed by keyauthorization = token. base64url (has256 (accountKey). digest (). The TXT value is computed by TXT = base64url (has256 (keyauthorization). digest (). Is this the way of calculation? I did not use CSR in authorization.

The challenge status I got from the day I redo was as follows::____________{

“Type”: “dns-01”.

“Status”: “pending”

“Url”: “https://acme-staging-v02.api.letsencrypt.org/acme/challenge/v2/207636/gId1tg==”

“Token”: “unW545EYkwHDUvtEk3Jb66u9gKy1BQFYsWWWWpolr3mk”

}

So far, the status of challenge has not changed.

I’ve been requesting POST / acme / chall / xxxxxxxxx, but the result is still the same, so I’m beginning to wonder if something went wrong in the previous steps.

1 Like
2

Can you show us the order URL related to that challenge URL?

I think you have the right idea (you need to do an authenticated POST to the challenge URL to respond to it), but it looks like you’ve made an implementation mistake somewhere along the way.

Even if you got the TXT value wrong, the challenge should become invalid, so no need to worry about that part yet.

3

this is order request info:
<type 'dict'>: {'nonce': 'oL3asROISUK0x-JZ2VSvRCNbnxyIvNXalPrkXHUqzJQ', 'alg': 'RS256', 'url': u'https://acme-staging-v02.api.letsencrypt.org/acme/new-order', 'kid': 'https://acme-staging-v02.api.letsencrypt.org/acme/acct/9629670'}
this is auth requst info:
{
"status": "pending",
"expires": "2019-06-24T07:09:55Z",
"identifiers": [
{
"type": "dns",
"value": "*.hzqp777.com"
}
],
"authorizations": [
"https://acme-staging-v02.api.letsencrypt.org/acme/authz/v2/207636"
],
"finalize": "https://acme-staging-v02.api.letsencrypt.org/acme/finalize/9629670/37662800"
}
Do you need a post auth request for https://acme-staging-v02.api.letsencrypt.org/acme/authz/v2/207636 after this step?

4

this order respons:
{
“status”: “pending”,
“expires”: “2019-06-24T07:09:55Z”,
“identifiers”: [
{
“type”: “dns”,
“value”: “*.hzqp777.com”
}
],
“authorizations”: [
“https://acme-staging-v02.api.letsencrypt.org/acme/authz/v2/207636”
],
“finalize”: “https://acme-staging-v02.api.letsencrypt.org/acme/finalize/9629670/37662800”
}

5

Once you have setup the TXT record with the key authorization value, you then need to send an “empty” POST to the challenge URL (https://acme-staging-v02.api.letsencrypt.org/acme/challenge/v2/207636/gId1tg==).

That process described here: https://tools.ietf.org/html/rfc8555#section-7.5.1

Then the challenge should proceed from the pending status to one of processing, valid, or invalid.

Have you already done that?

6

Yes, I did this step and finally got the theme of this post.

7

Alright.

It’s possible that your challenge became stuck in pending on the Let’s Encrypt side, but we will need to wait for a Let’s Encrypt staff member to take a look at it to verify.

If you can post your full ACME client code I could take a look at that, but otherwise we’ll need to wait.

8

'''

  • Created with IntelliJ Pycharm.
  • Description: ssl_cert_v2
  • Author 罗卫
  • User: devops
  • Date: 2019-06-11
  • Time: 上午9:49
  • Thank: John Hanley (myhelper,help,myutils)
    '''

import requests
import json
import myhelper
import os
from base.mylog import loglog

class ssl_cert_v2:

logs = loglog()
log = logs.logger
base_path = 'https://acme-staging-v02.api.letsencrypt.org/directory'

#request headers
headers = {
    'User-Agent': 'lw-ghy-acme-client/1.0',
    'Accept-Language': 'zh',
    'Content-Type':"application/jose+json"
}

#new resources path
nonec_path="newNonce"
nonec="Replay-Nonce"
account_path="newAccount"
order_path="newOrder"
account_order = 'orders?cursor=2>,rel="next"'
# old_order = 'orders?cursor=2>,rel="next"'
Authz_path="newAuthz"
revokeCert="revokeCert"
keyChange="keyChange"

AccountKeyFile = 'account.key'
EmailAddresses = ['mailto:123456@qq.com.com', 'mailto:987654@qq.com']

def get_directory(self):
    try:
        directorys = requests.get(self.base_path, headers=self.headers)
    except requests.exceptions.RequestException as error:
        self.log.error(error)

    if directorys.status_code < 200 or directorys.status_code >= 300:
        self.log.error('Error calling ACME endpoint:', directorys.reason)
    else:
        result = directorys.json()
        return result

def get_nonce(self,path):
    nonce = requests.head(path).headers[self.nonec]
    return nonce

def check_account_key_file(self):
    """ Verify that the Account Key File exists and prompt to create if it does not exist """
    if os.path.exists(self.AccountKeyFile) is not False:
        return True

    self.log.error('Error: File does not exist: {0}'.format(self.AccountKeyFile))

    if myhelper.Confirm('Create new account private key (y/n): ') is False:
        self.log.error('Cancelled')
        return False

    myhelper.create_rsa_private_key(self.AccountKeyFile)

    if os.path.exists(self.AccountKeyFile) is False:
        self.log.error('Error: File does not exist: {0}'.format(self.AccountKeyFile))
        return False

    return True

def data_packaging(self,payload,body_top):
    payload_b64 = myhelper.b64(json.dumps(payload).encode("utf8"))
    body_top_b64 = myhelper.b64(json.dumps(body_top).encode("utf8"))
    # Create the message digest (signature)
    data = "{0}.{1}".format(body_top_b64, payload_b64).encode("utf8")
    signature = myhelper.sign(data, self.AccountKeyFile)
    if signature == None:
        self.new_account()
        signature = myhelper.sign(data,self.AccountKeyFile)
    # Create the HTML request body
    jose = {"protected": body_top_b64,"payload": payload_b64,"signature": myhelper.b64(signature)}
    return jose

def new_account(self):
    if self.check_account_key_file() is False:
        self.make_account_private_key()
    new_accuount = self.get_directory()
    accuount_url = new_accuount[self.account_path]
    nonce = self.get_nonce(new_accuount[self.nonec_path])
    # Get the URL for the terms of service
    terms_service = new_accuount.get("meta", {}).get("termsOfService", "")
    self.log.info('Terms of Service:', terms_service)
    # Create the account request
    if terms_service != "":
        payload = {"termsOfServiceAgreed": True,"contact": self.EmailAddresses}
        self.log.info(payload)
    body_top = {"alg": "RS256","jwk": myhelper.get_jwk(self.AccountKeyFile),"url": accuount_url,"nonce": nonce}
    jose = self.data_packaging(payload,body_top)
    try:
        self.log.info('Calling endpoint:', accuount_url)
        resp = requests.post(accuount_url, json=jose, headers=self.headers)
    except requests.exceptions.RequestException as error:
        resp = error.response
        self.log.error(resp)
    except Exception as ex:
        self.log.error(ex)
    except BaseException as ex:
        self.log.error(ex)
    if resp.status_code < 200 or resp.status_code >= 300:
        self.log.info('Error calling ACME endpoint:', resp.reason)
        self.log.error('Status Code:', resp.status_code)
        myhelper.process_error_message(resp.text)
    if 'Location' in resp.headers:
        self.log.info('Account URL:', resp.headers['Location'])
    else:
        self.log.error('Error: Response headers did not contain the header "Location"')
        return "System error, please contact the system administrator!"

def get_account_url(self):
    dir_noce=self.get_directory()
    # Create the account request
    payload = {"termsOfServiceAgreed": True, "contact": self.EmailAddresses}
    nonce=self.get_nonce(dir_noce[self.nonec_path])
    body_top = {"alg": "RS256","jwk": myhelper.get_jwk(self.AccountKeyFile),"url": dir_noce[self.account_path],"nonce": nonce}
    jose = self.data_packaging(payload,body_top)
    # Make the ACME request
    try:
        resp = requests.post(dir_noce[self.account_path], json=jose, headers=self.headers)
    except requests.exceptions.RequestException as error:
        resp = error.response
        self.log.error(resp)
    except Exception as error:
        self.log.error(error)

    if resp.status_code < 200 or resp.status_code >= 300:
        print resp.reason
        self.log.error('Error calling ACME endpoint:%s'%resp.reason)
        self.log.error('Status Code:%s'%resp.status_code)
    else:
        if 'Location' in resp.headers:
            print
            self.log.info('Account URL:%s'%resp.headers['Location'])
            nonce = resp.headers[self.nonec]
            account_url = resp.headers['Location']
            return nonce, account_url
        else:
            self.log.info('INFO: Response headers did not contain the header "Location",Start new accounts.')
            self.new_account()
            self.get_account_url()
            self.log.info('INFO: The new account is created and returned to the account URL.')
            return "System error, please contact the system administrator!"

def get_account_info(self):
    """ Get the Account Information """
    accounts=self.get_account_url()
    # Create the account request
    if accounts != None:
        payload = {}
        body_top = {"alg": "RS256","kid": accounts[1],"nonce": accounts[0],"url": accounts[1]}
        jose = self.data_packaging(payload,body_top)
        try:
            resp = requests.post(accounts[1], json=jose, headers=self.headers)
        except requests.exceptions.RequestException as error:
            resp = error.response
            self.log.error(resp)
        except Exception as error:
            self.log.error(error)

        if resp.status_code < 200 or resp.status_code >= 300:
            self.log.error('Error calling ACME endpoint:%s'%resp.reason)
            self.log.error('Status Code:%s'%resp.status_code)
            return "System error, please contact the system administrator!"
        else:
            info = json.loads(resp.text)
            info["url"]=resp.url
            nonce = resp.headers[self.nonec]
            info["nonce"] = nonce
            return info
    else:
        return "System error, please contact the system administrator!"

def account_deactivate(self):
    """ Call ACME API to Deactivate Account """
    accounts=self.get_account_url()
    # Create the account request
    payload = {'status': 'deactivated'}
    body_top = {"alg": "RS256","kid": accounts[1],"nonce": accounts[0],"url": accounts[1]}
    jose = self.data_packaging(payload,body_top)
    host = accounts[1].split("//")[-1].split("/")[0].split('?')[0]
    self.headers['Host'] = host
    # Make the ACME request
    try:
        print('Calling endpoint:', accounts[1])
        resp = requests.post(accounts[1], json=jose, headers=self.headers)
    except requests.exceptions.RequestException as error:
        resp = error.response
        self.log.error(resp)
    except Exception as error:
        self.log.error(error)

    if resp.status_code < 200 or resp.status_code >= 300:
        self.log.error('Error calling ACME endpoint:', resp.reason)
        self.log.error('Status Code:%s'%resp.status_code)
        return "System error, please contact the system administrator!"
    else:
        info = json.loads(resp)
        return info

def account_update(self):
    accounts=self.get_account_url()
    # Create the account request
    payload = {"contact": self.EmailAddresses}
    body_top = {"alg": "RS256","kid": accounts[1],"nonce": accounts[0],"url": accounts[1]}
    host = accounts[1].split("//")[-1].split("/")[0].split('?')[0]
    self.headers['Host'] = host
    jose = self.data_packaging(payload,body_top)
    # Make the ACME request
    try:
        resp = requests.post(accounts[1], json=jose, headers=self.headers)
    except requests.exceptions.RequestException as error:
        resp = error.response
        self.log.error(resp)
    except Exception as error:
        self.log.error(error)
    if resp.status_code < 200 or resp.status_code >= 300:
        self.log.error('Error calling ACME endpoint:%s'%resp.reason)
        self.log.error('Status Code:%s'%resp.status_code)
        return "System error, please contact the system administrator!"
    else:
        info = json.loads(resp)
        return info

def new_order(self,domains):
    """ Request an SSL certificate from the ACME server """
    # domains = myhelper.get_domains_from_csr(csrfile)
    # Create the account request
    if type(domains) == list or type(domains) == tuple:
        accounts = self.get_account_url()
        dir = self.get_directory()
        order_url = dir[self.order_path]
        self.log.info("Request to the ACME server an order to validate domains.")
        payload = {"identifiers": [{"type": "dns", "value": domain} for domain in domains]}
        body_top = {"alg": "RS256","kid": accounts[1],"nonce": accounts[0],"url": dir[self.order_path]}
        jose = self.data_packaging(payload,body_top)
        # Make the ACME request
        try:
            resp = requests.post(order_url, json=jose, headers=self.headers)
        except requests.exceptions.RequestException as error:
            resp = error.response
            self.log.error(resp)
        except Exception as error:
            self.log.error(error)
        if resp.status_code < 200 or resp.status_code >= 300:
            self.log.error('Error calling ACME endpoint:%s'%resp.reason)
            self.log.error('Status Code:%s'%resp.status_code)
            return "System error, please contact the system administrator!"
        else:
            nonce = resp.headers[self.nonec]
            if resp.status_code == 201:
                order_location = resp.headers['Location']
                return order_location
    self.log.error( 'The type of domains must be "List" or "tuple".')
    return None

def old_order(self):
    accounts = self.get_account_url()
    self.log.info("Request to the ACME server an order to validate domains.")
    order_url = '%s/%s'%(accounts[1],self.account_order)
    payload = {}
    body_top = {"alg": "RS256", "kid": accounts[1], "nonce": accounts[0], "url": order_url}
    jose = self.data_packaging(payload, body_top)
    self.log.info("Request URL:%s"%order_url)
    try:
        resp = requests.post(order_url,json=jose,headers=self.headers)
    except requests.exceptions.RequestException as error:
        resp = error.response
        self.log.error(resp)
        return error
    except Exception as error:
        self.log.error(error)
        return error
    if resp.status_code < 200 or resp.status_code >= 300:
        self.log.error('Error calling ACME endpoint:%s' % resp.reason)
        self.log.error('Status Code:%s' % resp.status_code)
        return "System error, please contact the system administrator!"
    return resp

def get_auth(self,order_info):
    if order_info != None:
        try:
            resp = requests.get(order_info, headers=self.headers)
        except requests.exceptions.RequestException as error:
            resp = error.response
            self.log.error(resp)
            return  None
        except Exception as error:
            self.log.error(error)
            return None
        if resp.status_code < 200 or resp.status_code >= 300:
            self.log.error('Error calling ACME endpoint:%s'%resp.reason)
            self.log.error('Status Code:%s'%resp.status_code)
            self.log.error("System error, please contact the system administrator!")
        else:
            get_auth = json.loads(resp.text)
            return get_auth
    self.log.error("System error, please contact the system administrator!")
    return None

def get_challenges(self,auth_link):
    try:
        resp = requests.get(auth_link[0], headers=self.headers)
    except requests.exceptions.RequestException as error:
        resp = error.response
        self.log.error(resp)
        return None
    except Exception as error:
        self.log.error(error)
        return None
    if resp.status_code < 200 or resp.status_code >= 300:
        self.log.error('Error calling ACME endpoint:%s'%resp.reason)
        self.log.error('Status Code:%s'%resp.status_code)
        return "System error, please contact the system administrator!"
    get_challenges = json.loads(resp.text)
    return get_challenges

def join_Char(self,one,two):
    return "{0}.{1}".format(one, two)

def dns_auth(self,auth_info):
    LABLE = "_acem_challenge"
    challenge = self.get_challenges(auth_info["authorizations"])
    if challenge != None and challenge["identifier"] and challenge["challenges"]:
        domain_name = challenge["identifier"]["value"]
        token=challenge["challenges"][0]["token"]
        new_accuount = self.get_directory()
        nonce = self.get_nonce(new_accuount[self.nonec_path])
        payload = {}
        account_key = myhelper.get_jwk(self.AccountKeyFile)
        account_url = self.get_account_url()[1]
        keyAuthorization = self.join_Char(token,myhelper.b64(myhelper.JWK_Thumbprint(account_key)))
        body_top = {"alg": "RS256","kid":account_url,"url": challenge["challenges"][0]["url"],"nonce": nonce}
        jose = self.data_packaging(payload,body_top)
        try:
            resp = requests.post(challenge["challenges"][0]["url"],json=jose,headers=self.headers)
        except requests.exceptions.RequestException as error:
            resp = error.response
            self.log.error(resp)
            return None
        except Exception as error:
            self.log.error(error)
            return None
        if resp.status_code < 200 or resp.status_code >= 300:
            self.log.error('Error calling ACME endpoint:%s' % resp.reason)
            self.log.error('Status Code:%s' % resp.status_code)
            self.log.error("[ERROR] All info: %s"%json.dumps(resp.text))
            return "System error, please contact the system administrator!"
        TXT = myhelper.b64(myhelper.hash_256_digest(keyAuthorization))
        name = self.join_Char(LABLE, domain_name)
        return ["DNS parse name: %s type: TXT value: %s "%(name,TXT),auth_info["authorizations"][0],challenge["challenges"][0]["url"]]
    self.log.error("[Error]: DNS auth error, data request exception.")
    return None

def dns_validation(self,challenge):
    new_accuount = self.get_directory()
    nonce = self.get_nonce(new_accuount[self.nonec_path])
    payload = {}
    account_url = self.get_account_url()[1]
    body_top = {"alg": "RS256","kid":account_url,"url": challenge,"nonce": nonce}
    jose = self.data_packaging(payload,body_top)
    try:
        resp = requests.post(challenge,json=jose,headers=self.headers)
    except requests.exceptions.RequestException as error:
        resp = error.response
        self.log.error(resp)
        return None
    except Exception as error:
        self.log.error(error)
        return None
    if resp.status_code < 200 or resp.status_code >= 300:
        self.log.error('Error calling ACME endpoint:%s' % resp.reason)
        self.log.error('Status Code:%s' % resp.status_code)
        self.log.error("[ERROR] All info: %s"%json.dumps(resp.text))
        return None
    if resp.status_code == 201:
        order_location = resp.headers['Location']
        return order_location
    return json.loads(resp.text)

def dns_challenge(self,challenge_link):
    if challenge_link != None:
        new_accuount = self.get_directory()
        nonce = self.get_nonce(new_accuount[self.nonec_path])
        payload = {}
        account_url = self.get_account_url()[1]
        body_top = {"alg": "RS256","kid":account_url,"url": challenge_link["url"],"nonce": nonce}
        jose = self.data_packaging(payload,body_top)
        try:
            resp = requests.post(challenge_link["url"],json=jose,headers=self.headers)
        except requests.exceptions.RequestException as error:
            resp = error.response
            self.log.error(resp)
            return None
        except Exception as error:
            self.log.error(error)
            return None
        if resp.status_code < 200 or resp.status_code >= 300:
            self.log.error('Error calling ACME endpoint:%s' % resp.reason)
            self.log.error('Status Code:%s' % resp.status_code)
            self.log.error("[ERROR] All info: %s"%json.dumps(resp.text))
            return "本次申请状态已失效，请重新输入域名点击提交按钮"
        if resp.status_code == 201:
            order_location = resp.headers['Location']
            return order_location
        return resp.text
    return "System error, please contact the system administrator!"

def get_cert(self):
    pass

def revokecert(self):
    pass
def keychange(self):
    pass
9

this client Source code。There may be places where naming is not a special specification

10

The old order method has been abandoned and is intended to implement account order query, but it is not implemented in acme api.

11

I’m not able to run this example because of missing dependencies (myhelper, base.mylog).

But in general, you need to keep polling the challenge URL after responding to the challenge.

In dns_auth & dns_validation & dns_challenge, you are only checking the status once.

It may take more time for the challenge to update. Clients like Certbot will keep checking the challenge status by re-fetching the challenge URL, once per second for 30 seconds, before giving up.

If you can package up your code with its dependencies so that it can be run in a standalone way (along with a sample main program), I can help you debug it further.

1 Like
12

I put this project on gitee, which is git address: https://gitee.com/luowei_lv/autossl.git

13

Thanks.

I have downloaded and run your application, using your /applyssl endpoint.

I believe the problem is on Let’s Encrypt’s side.

At the moment, all orders to acme-staging-v02.api.letsencrypt.org that contain wildcard identifiers are resulting in challenges that permanently stay in pending status. I have replicated the same scenario with Certbot. @lestaff

Unfortunately there’s not a lot you can do - you can try using https://github.com/letsencrypt/pebble to develop your client instead of using the Let’s Encrypt staging environment.

When changing the ACME server in your application to the production one, it works as expected (except you need to poll the challenges as I mentioned before …).

2 Likes
14

https://github.com/letsencrypt/pebble， Pebble is NOT INTENDED FOR PRODUCTION USE . Pebble is for testing only .

By design Pebble will drop all of its state between invocations and will randomize keys/certificates used for issuance.

15

Yes, that’s correct. Pebble can provide a test ACME server for you to develop your project against.

Once your project works against Pebble on localhost, you can change it to use Let’s Encrypt API server.

At least, this way you can continue testing your client, while the issue with Let’s Encrypt is outstanding.

2 Likes
16

Okay, I’ll try to be a local test. Once I get into the pre-production test deployment, I’ll change a domain name to do this test.

1 Like
17

Hi,
Same here, I’m using the v2 staging URL(“https://acme-staging-v02.api.letsencrypt.org/directory”), and Dns-01 challenge stays “pending”…

Yesterday everything worked well, so I believe the problem is on Let’s Encrypt’s side.

1 Like
18

Thank you, this should be the problem, but now I have to take other ways and solutions to continue. But I haven't found a way yet.

19

Now when using pebble local test service, it appears at startup: 2019/06/18:18:27:35 http: TLS handshake error from [::1]:47864: remote error: tls: unknown certificate authority, which I am not sure what caused the failure of the client to link. Can you help me?

20

Looking into it. (Free bonus text to meet the min. post length)

1 Like