From df2a6192f532779e4a9faf74e35df8f74b7085ad Mon Sep 17 00:00:00 2001 From: Dmitriy Fofanov Date: Wed, 25 Feb 2026 08:06:28 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=BE=20=D0=BE=D0=BF=D1=80=D0=B5=D0=B4=D0=B5=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D0=B5=20=D0=B0=D0=BA=D1=82=D1=83=D0=B0=D0=BB=D1=8C?= =?UTF-8?q?=D0=BD=D0=BE=D0=B9=20=D0=B4=D0=B8=D1=80=D0=B5=D0=BA=D1=82=D0=BE?= =?UTF-8?q?=D1=80=D0=B8=D0=B8=20=D1=81=D0=B5=D1=80=D1=82=D0=B8=D1=84=D0=B8?= =?UTF-8?q?=D0=BA=D0=B0=D1=82=D0=B0=20=D0=B8=20=D1=83=D0=BB=D1=83=D1=87?= =?UTF-8?q?=D1=88=D0=B5=D0=BD=D0=B0=20=D0=BE=D0=B1=D1=80=D0=B0=D0=B1=D0=BE?= =?UTF-8?q?=D1=82=D0=BA=D0=B0=20=D0=B8=D0=BC=D0=B5=D0=BD=20=D1=81=D0=B5?= =?UTF-8?q?=D1=80=D1=82=D0=B8=D1=84=D0=B8=D0=BA=D0=B0=D1=82=D0=BE=D0=B2=20?= =?UTF-8?q?=D0=B2=20=D0=BA=D0=BB=D0=B0=D1=81=D1=81=D0=B5=20LetsEncryptMana?= =?UTF-8?q?ger?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- letsencrypt_regru_api.py | 204 ++++++++++++++++++++++++++++++++++----- 1 file changed, 182 insertions(+), 22 deletions(-) diff --git a/letsencrypt_regru_api.py b/letsencrypt_regru_api.py index 9d2647d..9a9b8ad 100644 --- a/letsencrypt_regru_api.py +++ b/letsencrypt_regru_api.py @@ -1071,7 +1071,65 @@ class LetsEncryptManager: self.logger = logger self.domain = config["domain"] self.email = config["email"] - self.cert_dir = os.path.join(config["cert_dir"], self.domain) + self.cert_name = get_primary_cert_name(config.get("cert_domains", []), self.domain) + self.cert_dir = self._resolve_live_cert_dir() + + def _resolve_live_cert_dir(self) -> str: + """ + Определяет актуальную директорию сертификата в /etc/letsencrypt/live. + + Почему это важно: + - cert_name не должен быть wildcard ("*.example.com") + - certbot может использовать suffix вида "-0001" + - в старых конфигурациях directory могла не совпадать с self.domain + """ + live_root = self.config["cert_dir"] + + # 1) Явные кандидаты + candidates = [] + for candidate in [self.cert_name, self.domain, self.domain.replace("*.", "")]: + if candidate and candidate not in candidates: + candidates.append(candidate) + + for candidate in candidates: + candidate_dir = os.path.join(live_root, candidate) + if os.path.exists(candidate_dir): + return candidate_dir + + # 1.1) Линии certbot с суффиксом -0001/-0002 + try: + if os.path.exists(live_root): + prefixed = [ + name for name in os.listdir(live_root) + if name.startswith(f"{self.cert_name}-") and os.path.isdir(os.path.join(live_root, name)) + ] + if prefixed: + prefixed.sort() + return os.path.join(live_root, prefixed[-1]) + except Exception: + pass + + # 2) Пытаемся получить путь напрямую из certbot --cert-name + try: + result = subprocess.run( + ["certbot", "certificates", "--cert-name", self.cert_name], + capture_output=True, + text=True, + check=False, + ) + output = (result.stdout or "") + "\n" + (result.stderr or "") + for line in output.splitlines(): + line = line.strip() + if line.startswith("Certificate Path:"): + cert_path = line.split("Certificate Path:", 1)[1].strip() + cert_dir = os.path.dirname(cert_path) + if cert_dir: + return cert_dir + except Exception: + pass + + # 3) Fallback для первого выпуска + return os.path.join(live_root, self.cert_name) def check_certbot_installed(self) -> bool: """ @@ -1357,6 +1415,62 @@ class LetsEncryptManager: self.logger.info(f"Домен: {registrable_domain}, Поддомен: {subdomain}") return self.api.remove_txt_record(registrable_domain, subdomain, validation_token) + + def _get_authoritative_nameservers(self, domain: str) -> List[str]: + """Возвращает список authoritative NS для домена.""" + nameservers: List[str] = [] + + # 1) dig +short NS domain + try: + result = subprocess.run( + ["dig", "+short", "NS", domain], + capture_output=True, + text=True, + timeout=10 + ) + for line in result.stdout.splitlines(): + ns = line.strip().rstrip('.') + if ns and ns not in nameservers: + nameservers.append(ns) + except Exception: + pass + + # 2) fallback: nslookup -type=NS domain + if not nameservers: + try: + result = subprocess.run( + ["nslookup", "-type=NS", domain], + capture_output=True, + text=True, + timeout=10 + ) + for line in result.stdout.splitlines(): + low = line.lower() + if "nameserver" in low and "=" in line: + ns = line.split("=", 1)[1].strip().rstrip('.') + if ns and ns not in nameservers: + nameservers.append(ns) + except Exception: + pass + + return nameservers + + def _query_txt_nslookup(self, full_domain: str, nameserver: Optional[str] = None) -> str: + """Запрашивает TXT через nslookup. Если nameserver задан — запрос к нему.""" + cmd = ["nslookup", "-type=TXT", full_domain] + if nameserver: + cmd.append(nameserver) + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=12 + ) + return result.stdout or "" + except Exception: + return "" def verify_dns_record_external(self, domain: str, subdomain: str, expected_value: str) -> bool: """ @@ -1380,26 +1494,52 @@ class LetsEncryptManager: self.logger.info(f" Ожидаемое значение: {expected_value[:30]}...") self.logger.info(f" Попыток: {attempts}, интервал: {interval} сек") self.logger.info("") + + auth_ns = self._get_authoritative_nameservers(domain) + if auth_ns: + self.logger.info(f" Авторитетные NS: {', '.join(auth_ns)}") + else: + self.logger.warning(" Не удалось определить authoritative NS, используем публичные DNS") + + public_resolvers = ["1.1.1.1", "8.8.8.8", "9.9.9.9"] for attempt in range(attempts): try: - # Используем nslookup или dig через subprocess - result = subprocess.run( - ["nslookup", "-type=TXT", full_domain], - capture_output=True, - text=True, - timeout=10 - ) - - if expected_value in result.stdout: - self.logger.info(f" ✅ Попытка {attempt + 1}/{attempts}: DNS запись НАЙДЕНА!") - # Показываем найденную запись - for line in result.stdout.split('\n'): + # 1) Проверка через authoritative NS (самый точный источник) + for ns in auth_ns: + ns_output = self._query_txt_nslookup(full_domain, ns) + if expected_value in ns_output: + self.logger.info(f" ✅ Попытка {attempt + 1}/{attempts}: TXT найден на authoritative NS {ns}") + for line in ns_output.split('\n'): + if 'text =' in line.lower() or expected_value[:20] in line: + self.logger.info(f" {line.strip()}") + return True + + # 2) Проверка через публичные резолверы + found_public = False + for resolver in public_resolvers: + res_output = self._query_txt_nslookup(full_domain, resolver) + if expected_value in res_output: + self.logger.info(f" ✅ Попытка {attempt + 1}/{attempts}: TXT найден через публичный DNS {resolver}") + for line in res_output.split('\n'): + if 'text =' in line.lower() or expected_value[:20] in line: + self.logger.info(f" {line.strip()}") + found_public = True + break + + if found_public: + return True + + # 3) Fallback на системный resolver + default_output = self._query_txt_nslookup(full_domain) + if expected_value in default_output: + self.logger.info(f" ✅ Попытка {attempt + 1}/{attempts}: TXT найден через системный DNS") + for line in default_output.split('\n'): if 'text =' in line.lower() or expected_value[:20] in line: self.logger.info(f" {line.strip()}") return True - else: - self.logger.info(f" ⏳ Попытка {attempt + 1}/{attempts}: DNS запись не найдена, ждём...") + + self.logger.info(f" ⏳ Попытка {attempt + 1}/{attempts}: DNS запись не найдена, ждём...") except Exception as e: self.logger.info(f" ⚠️ Попытка {attempt + 1}/{attempts}: Ошибка nslookup - {e}") @@ -1510,6 +1650,7 @@ class LetsEncryptManager: "certbot", "certonly", "--manual", "--preferred-challenges", "dns", + "--cert-name", self.cert_name, "--manual-auth-hook", auth_hook_script.name, "--manual-cleanup-hook", cleanup_hook_script.name, "--email", self.email, @@ -1532,6 +1673,7 @@ class LetsEncryptManager: self.logger.info("ЗАПУСК CERTBOT") self.logger.info("=" * 80) self.logger.info(f"Режим: {'STAGING (тестовый)' if staging else 'PRODUCTION (боевой)'}") + self.logger.info(f"Cert name: {self.cert_name}") self.logger.info(f"Команда: {' '.join(cmd)}") self.logger.info(f"Python: {sys.executable}") self.logger.info(f"Скрипт: {os.path.abspath(__file__)}") @@ -1661,19 +1803,15 @@ class LetsEncryptManager: cleanup_hook_script.close() os.chmod(cleanup_hook_script.name, 0o755) - # Используем cert_name — первый домен из SAN группы (или self.domain) - cert_domains = self.config.get("cert_domains") - cert_name = cert_domains[0] if cert_domains and isinstance(cert_domains, list) else self.domain - cmd = [ "certbot", "renew", - "--cert-name", cert_name, + "--cert-name", self.cert_name, "--manual-auth-hook", auth_hook_script.name, "--manual-cleanup-hook", cleanup_hook_script.name, "--non-interactive", ] - self.logger.info(f"Домен (cert-name): {cert_name}") + self.logger.info(f"Домен (cert-name): {self.cert_name}") self.logger.info(f"Конфигурация: {config_path}") self.logger.info(f"Auth hook: {sys.executable} {os.path.abspath(__file__)} --config {config_path} --auth-hook") self.logger.info(f"Cleanup hook: {sys.executable} {os.path.abspath(__file__)} --config {config_path} --cleanup-hook") @@ -1952,6 +2090,27 @@ def get_domain_config(base_config: Dict, domain: str) -> Dict: return domain_config +def get_primary_cert_name(cert_domains: List[str], fallback_domain: str) -> str: + """ + Возвращает стабильное имя сертификата для certbot (--cert-name). + + Приоритет: + 1) первый НЕ-wildcard домен из cert_domains + 2) первый домен из cert_domains без префикса "*." + 3) fallback_domain без префикса "*." + """ + if isinstance(cert_domains, list): + for domain in cert_domains: + if isinstance(domain, str) and domain and not domain.startswith("*."): + return domain.strip() + + for domain in cert_domains: + if isinstance(domain, str) and domain: + return domain.replace("*.", "").strip() + + return fallback_domain.replace("*.", "").strip() + + def get_cert_group_config(base_config: Dict, cert_group: List[str]) -> Dict: """ Создаёт копию конфигурации для группы SAN-доменов (один сертификат). @@ -1964,7 +2123,8 @@ def get_cert_group_config(base_config: Dict, cert_group: List[str]) -> Dict: Конфигурация с domain = первый домен, cert_domains = все домены группы """ group_config = base_config.copy() - group_config["domain"] = cert_group[0] # primary domain (cert-name) + cert_name = get_primary_cert_name(cert_group, base_config.get("domain", "")) + group_config["domain"] = cert_name group_config["cert_domains"] = cert_group # все SAN домены return group_config