From 170c48e2577484d214d83b791d60ba0a1d0f9dd9 Mon Sep 17 00:00:00 2001
From: kappa <kappa@inouter.com>
Date: Sun, 8 Feb 2026 20:34:57 +0900
Subject: [PATCH] Detect subdomains structurally to skip wildcard entries
 without certs

Add CUSTOM_TLDS config (HAPROXY_CUSTOM_TLDS env, default: "it.com")
and _get_base_domain() for eTLD+1 detection. _check_subdomain now uses
three layers: registered domains, certificate domains, and structural
analysis. This ensures nocodb.inouter.com never gets a *.nocodb wildcard
entry even when inouter.com has no cert or registration.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 haproxy_mcp/config.py        |  6 +++++
 haproxy_mcp/tools/domains.py | 51 +++++++++++++++++++++++++++++++-----
 2 files changed, 50 insertions(+), 7 deletions(-)

diff --git a/haproxy_mcp/config.py b/haproxy_mcp/config.py
index 01470b1..97bc971 100644
--- a/haproxy_mcp/config.py
+++ b/haproxy_mcp/config.py
@@ -39,6 +39,12 @@ CERTS_DIR: str = os.getenv("HAPROXY_CERTS_DIR", "/opt/haproxy/certs")
 CERTS_DIR_CONTAINER: str = os.getenv("HAPROXY_CERTS_DIR_CONTAINER", "/etc/haproxy/certs")
 ACME_HOME: str = os.getenv("ACME_HOME", os.path.expanduser("~/.acme.sh"))
 
+# Custom multi-part TLDs (e.g., "it.com" treated as a TLD so "anvil.it.com" is a base domain)
+# Comma-separated list via env var, or default
+CUSTOM_TLDS: frozenset[str] = frozenset(
+    t.strip() for t in os.getenv("HAPROXY_CUSTOM_TLDS", "it.com").split(",") if t.strip()
+)
+
 # Pool configuration
 POOL_COUNT: int = int(os.getenv("HAPROXY_POOL_COUNT", "100"))
 MAX_SLOTS: int = int(os.getenv("HAPROXY_MAX_SLOTS", "10"))
diff --git a/haproxy_mcp/tools/domains.py b/haproxy_mcp/tools/domains.py
index 2ca657e..ab43666 100644
--- a/haproxy_mcp/tools/domains.py
+++ b/haproxy_mcp/tools/domains.py
@@ -12,6 +12,7 @@ from ..config import (
     MAX_SLOTS,
     SUBPROCESS_TIMEOUT,
     CERTS_DIR,
+    CUSTOM_TLDS,
     REMOTE_MODE,
     logger,
 )
@@ -37,13 +38,43 @@ from ..db import db_load_certs
 from ..utils import parse_servers_state, disable_server_slot
 
 
-def _check_subdomain(domain: str, registered_domains: set[str]) -> tuple[bool, Optional[str]]:
-    """Check if a domain is a subdomain of an existing registered domain or certificate domain.
+def _get_base_domain(domain: str) -> Optional[str]:
+    """Get the base domain (eTLD+1) considering custom multi-part TLDs.
 
-    For example, vault.anvil.it.com is a subdomain if anvil.it.com exists.
-    nocodb.inouter.com is a subdomain if inouter.com has a certificate.
-    Subdomains should not have wildcard entries added to avoid conflicts,
-    because wildcard certs (*.example.com) only cover one level deep.
+    Examples (with CUSTOM_TLDS={"it.com"}):
+        inouter.com -> inouter.com (base domain itself)
+        nocodb.inouter.com -> inouter.com
+        anvil.it.com -> anvil.it.com (base domain, it.com is TLD)
+        gitea.anvil.it.com -> anvil.it.com
+
+    Returns:
+        The base domain, or None if the domain is a TLD itself.
+    """
+    parts = domain.split(".")
+
+    # Check custom multi-part TLDs first (e.g., it.com)
+    for tld in CUSTOM_TLDS:
+        tld_parts = tld.split(".")
+        if len(parts) > len(tld_parts) and domain.endswith("." + tld):
+            return ".".join(parts[-(len(tld_parts) + 1):])
+
+    # Standard single-part TLD (e.g., .com, .net, .org)
+    if len(parts) >= 2:
+        return ".".join(parts[-2:])
+
+    return None
+
+
+def _check_subdomain(domain: str, registered_domains: set[str]) -> tuple[bool, Optional[str]]:
+    """Check if a domain is a subdomain using structural analysis and known domains.
+
+    Uses three layers of detection:
+    1. Registered domains: vault.anvil.it.com is a subdomain if anvil.it.com is registered.
+    2. Certificate domains: nocodb.inouter.com is a subdomain if inouter.com has a cert.
+    3. Structural analysis: nocodb.inouter.com is deeper than eTLD+1 (inouter.com).
+
+    Wildcard entries are skipped for subdomains because wildcard certs
+    (*.example.com) only cover one level deep.
 
     Args:
         domain: Domain name to check (e.g., "api.example.com").
@@ -52,7 +83,7 @@ def _check_subdomain(domain: str, registered_domains: set[str]) -> tuple[bool, O
     Returns:
         Tuple of (is_subdomain, parent_domain or None).
     """
-    # Combine registered domains and certificate domains as known base domains
+    # Check against registered domains and certificate domains
     cert_domains = set(db_load_certs())
     known_domains = registered_domains | cert_domains
 
@@ -61,6 +92,12 @@ def _check_subdomain(domain: str, registered_domains: set[str]) -> tuple[bool, O
         candidate = ".".join(parts[i:])
         if candidate in known_domains:
             return True, candidate
+
+    # Structural analysis: if domain is deeper than its base domain, it's a subdomain
+    base = _get_base_domain(domain)
+    if base and domain != base:
+        return True, base
+
     return False, None