Добавлено определение актуальной директории сертификата и улучшена обработка имен сертификатов в классе LetsEncryptManager

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