Provide Plesk support together with Certbot

certbotplesklib.py

import os
import requests
import xml.etree.ElementTree as xml

site_id = 'PleskID'
username = 'PleskUsername'
password = 'PleskPassword'
plesk_node = 'PleskNode'

#
# remove domain root from domain
#
def extract_sub_domain(domain):
  parts = domain.split('.')
  depth = len(parts)
  record_name = ''

  if depth == 2:
    return ''

  for i in range(0, (depth - 2)):
    if i <> 0:
      record_name += '.'
    record_name += parts[i]

  return record_name

#
# generating the DNS Add request
#
def build_dns_add_record(record_type, record_name, record_value):
  packet = xml.Element('packet')
  dns = xml.SubElement(packet, 'dns')
  rec = xml.SubElement(dns, 'add_rec')
  xml.SubElement(rec, 'site-id').text = site_id
  xml.SubElement(rec, 'type').text = record_type
  xml.SubElement(rec, 'host').text = record_name
  xml.SubElement(rec, 'value').text = record_value

  return packet

#
# parse DNS Add response
#
def parse_dns_add_response(body):
  packet = xml.fromstring(body)
  result = packet.find('dns/add_rec/result')

  if result.find('status').text != 'ok':
    raise Exception('The DNS record creation has failed.', body)

  return result.find('id').text

#
# generating the DNS Delete request
#
def build_dns_delete_record(dns_record_id):
  packet = xml.Element('packet')
  dns = xml.SubElement(packet, 'dns')
  rec = xml.SubElement(dns, 'del_rec')
  filter = xml.SubElement(rec, 'filter')
  xml.SubElement(filter, 'id').text = dns_record_id

  return packet

#
# parse the DNS Delete response
#
def parse_dns_delete_response(body):
  packet = xml.fromstring(body)
  result = packet.find('dns/del_rec/result')

  return result.find('status').text == 'ok'

#
# sending the request
#
def send_request(packet):
  base_url = 'https://%s:8443/enterprise/control/agent.php' % (plesk_node)
  headers = {'HTTP_AUTH_LOGIN':username,'HTTP_AUTH_PASSWD':password,'Content-Type':'application/xml'}

  return requests.post(url = base_url, headers = headers, data = xml.tostring(packet))

#
# ACME Authenticate hook
#
def acme_auth(domain, verification_token):
  directory = '/tmp/%s' % (domain)
  path = '%s/verification.xml' % (directory)

  record_name = '_acme-challenge'
  sub_domain = extract_sub_domain(domain)

  if sub_domain <> '':
    record_name += '.'

  record_name += sub_domain
  
  packet = build_dns_add_record('TXT', '_acme-challenge', verification_token)
  response = send_request(packet)
  record_id = parse_dns_add_response(response.text)

  # keep record creation response in order to beeing able to delete it on cleanup stage
  if not os.path.exists(directory):
    os.makedirs(directory)

  file = open(path, 'w+')
  file.write(response.text)
  file.close()

#
# ACME Cleanup hook
#
def acme_cleanup(domain):
  path = '/tmp/%s/verification.xml' % (domain)

  # open the DNS Creation response file
  file = open(path, 'r')
  response = file.read()
  file.close()

  # parse it and retrieve the created record ID
  record_id = parse_dns_add_response(response)

  # build a packet and send a DELETE request
  packet = build_dns_delete_record(record_id)
  response = send_request(packet)
  if not parse_dns_delete_response(response.text):
    raise Exception('Unable to delete the verification record !', response.text)

  os.remove(path)