Vous ne pouvez pas sélectionner plus de 25 sujets Les noms de sujets doivent commencer par une lettre ou un nombre, peuvent contenir des tirets ('-') et peuvent comporter jusqu'à 35 caractères.

PdnsApiAuthenticator.py 4.4KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  1. #!/usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. import json
  4. import logging
  5. import time
  6. from certbot import errors
  7. from certbot_pdns.pdnsapi import PdnsApi
  8. logger = logging.getLogger(__name__)
  9. class PdnsApiAuthenticator:
  10. api = None
  11. zones = None
  12. axfr_time = None
  13. def find_best_matching_zone(self, domain):
  14. if domain is None or domain == "":
  15. return None
  16. for zone in self.zones:
  17. if zone['name'] == domain + ".":
  18. return zone
  19. return self.find_best_matching_zone(domain[domain.index(".") + 1:]) if "." in domain else None
  20. def find_soa(self, zone):
  21. for rrset in zone["rrsets"]:
  22. if rrset["type"] == "SOA":
  23. return rrset, rrset["records"][0]
  24. return None
  25. def flush_zone(self, zone_name):
  26. res = self.api.flush_zone_cache(zone_name)
  27. if res is None or "result" not in res or res["result"] != "Flushed cache.":
  28. raise errors.PluginError("Bad return from PDNS API when flushing cache: %s" % res)
  29. def notify_zone(self, zone_name):
  30. res = self.api.notify_zone(zone_name)
  31. if res is None or "result" not in res or res["result"] != "Notification queued":
  32. raise errors.PluginError("Bad return from PDNS API when notifying: %s" % res)
  33. def update_soa(self, zone_name):
  34. zone = self.api.get_zone(zone_name)
  35. if zone is None or "error" in zone:
  36. raise errors.PluginError("Bad return from PDNS API when getting zone %s: %s" % (zone_name, zone))
  37. rrset, soa = self.find_soa(zone)
  38. split = soa["content"].split(" ")
  39. split[2] = str(int(split[2]) + 1)
  40. soa["content"] = ' '.join(split)
  41. res = self.api.replace_record(zone_name, zone_name, rrset["type"], rrset["ttl"], soa["content"],
  42. soa["disabled"], False)
  43. if res is not None:
  44. raise errors.PluginError("Bad return from PDNS API when updating SOA: %s" % res)
  45. def prepare(self, conf_path):
  46. self.api = PdnsApi()
  47. with open(conf_path) as f:
  48. config = json.load(f)
  49. self.api.set_api_key(config["api-key"])
  50. self.api.set_base_url(config["base-url"])
  51. self.axfr_time = config["axfr-time"]
  52. # check if additional parameters are set before trying to assign them to ensure backwards compatibility
  53. if "verify-cert" in config:
  54. self.api.set_verify_cert(config["verify-cert"])
  55. if "http-auth" in config:
  56. self.api.set_http_auth(config["http-auth"])
  57. self.zones = self.api.list_zones()
  58. # print(self.zones)
  59. # raw_input('Press <ENTER> to continue')
  60. if self.zones is None or "error" in self.zones:
  61. raise errors.PluginError("Could not list zones %s" % self.zones)
  62. def perform_single(self, achall, response, validation):
  63. domain = achall.domain
  64. token = validation.encode()
  65. zone = self.find_best_matching_zone(domain)
  66. if zone is None:
  67. raise errors.PluginError("Could not find zone for %s" % domain)
  68. logger.debug("Found zone %s for domain %s" % (zone["name"], domain))
  69. res = self.api.replace_record(zone["name"], "_acme-challenge." + domain + ".", "TXT", 1, "\"" + token.decode('utf-8') + "\"", False, False)
  70. if res is not None:
  71. raise errors.PluginError("Bad return from PDNS API when adding record: %s" % res)
  72. return response
  73. def perform_notify(self, zone):
  74. logger.info("Notifying zone %s..." % zone["name"])
  75. self.update_soa(zone["name"])
  76. self.flush_zone(zone["name"])
  77. self.notify_zone(zone["name"])
  78. def wait_for_propagation(self, achalls):
  79. # TODO search zones authoritative servers and check for TXT record on each of them
  80. # raw_input('Press <ENTER> to continue')
  81. logger.info("Waiting %i seconds..." % self.axfr_time)
  82. time.sleep(self.axfr_time)
  83. def cleanup(self, achall):
  84. domain = achall.domain
  85. zone = self.find_best_matching_zone(domain)
  86. if zone is None:
  87. return
  88. res = self.api.delete_record(zone["name"], "_acme-challenge." + domain + ".", "TXT", 1, None, False, False)
  89. if res is not None:
  90. raise errors.PluginError("Bad return from PDNS API when deleting record: %s" % res)
  91. self.update_soa(zone["name"])
  92. self.flush_zone(zone["name"])
  93. self.notify_zone(zone["name"])