Merge pull request #180 from rtCamp/feat/dns-challenge-certbot

Add Cloudflare Let's Encrypt DNS challenge support

frappe_manager/commands.py

@@ -25,7 +25,7 @@
 from frappe_manager.migration_manager.migration_executor import MigrationExecutor
 from frappe_manager.site_manager.site import Bench
 from frappe_manager.site_manager.workers_manager.SiteWorker import BenchWorkers
-from frappe_manager.ssl_manager import SUPPORTED_SSL_TYPES
+from frappe_manager.ssl_manager import LETSENCRYPT_PREFERRED_CHALLENGE, SUPPORTED_SSL_TYPES
 from frappe_manager.ssl_manager.certificate import SSLCertificate
 from frappe_manager.ssl_manager.letsencrypt_certificate import LetsencryptSSLCertificate
 from frappe_manager.utils.callbacks import (
@@ -156,6 +156,14 @@ def create(
     environment: Annotated[
         FMBenchEnvType, typer.Option("--environment", "--env", help="Select bench environment type.")
     ] = FMBenchEnvType.dev,
+    letsencrypt_preferred_challenge: Annotated[
+        Optional[LETSENCRYPT_PREFERRED_CHALLENGE],
+        typer.Option(help="Select preferred letsencrypt challenge.", show_default=False),
+    ] = None,
+    letsencrypt_email: Annotated[
+        Optional[str],
+        typer.Option(help="Specify email for letsencrypt", show_default=False),
+    ] = None,
     developer_mode: Annotated[
         EnableDisableOptionsEnum, typer.Option(help="Toggle frappe developer mode.")
     ] = EnableDisableOptionsEnum.disable,
@@ -205,18 +213,36 @@ def create(
     bench_config_path = bench_path / CLI_BENCH_CONFIG_FILE_NAME
 
     if ssl == SUPPORTED_SSL_TYPES.le:
-        if fm_config_manager.le_email == '[email protected]':
-            email = richprint.prompt_ask(prompt='Please enter [bold][green]email[/bold][/green] for Let\'s Encrypt')
+        if not letsencrypt_preferred_challenge:
+            if fm_config_manager.letsencrypt.exists:
+                if letsencrypt_preferred_challenge is None:
+                    letsencrypt_preferred_challenge = LETSENCRYPT_PREFERRED_CHALLENGE.dns01
+
+            if not letsencrypt_preferred_challenge:
+                letsencrypt_preferred_challenge = LETSENCRYPT_PREFERRED_CHALLENGE.http01
+
+        if fm_config_manager.letsencrypt.email == '[email protected]' or fm_config_manager.letsencrypt.email is None:
+            if not letsencrypt_email:
+                richprint.stop()
+                raise typer.BadParameter("No email provided, required by certbot.", param_hint='--letsencrypt-email')
+            else:
+                email = letsencrypt_email
+
             validate_email(email, check_deliverability=False)
-            fm_config_manager.le_email = email
-            fm_config_manager.export_to_toml()
-            richprint.print("Let's Encrypt email saved to configuration. It will be used automatically from now on.")
         else:
-            richprint.print("Using Let's Encrypt email from configuration.")
-            email = fm_config_manager.le_email
-            fm_config_manager.export_to_toml()
-
-        ssl_certificate = LetsencryptSSLCertificate(domain=benchname, ssl_type=ssl, email=email)
+            richprint.print(
+                "Defaulting to Let's Encrypt email from [blue]fm_config.toml[/blue] since [blue]'--letsencrypt-email'[/blue] is not given."
+            )
+            email = fm_config_manager.letsencrypt.email
+
+        ssl_certificate = LetsencryptSSLCertificate(
+            domain=benchname,
+            ssl_type=ssl,
+            email=email,
+            preferred_challenge=letsencrypt_preferred_challenge,
+            api_key=fm_config_manager.letsencrypt.api_key,
+            api_token=fm_config_manager.letsencrypt.api_token,
+        )
 
     elif ssl == SUPPORTED_SSL_TYPES.none:
         ssl_certificate = SSLCertificate(domain=benchname, ssl_type=ssl)
@@ -483,9 +509,17 @@ def update(
         Optional[EnableDisableOptionsEnum],
         typer.Option("--admin-tools", help="Toggle admin-tools.", show_default=False),
     ] = None,
+    letsencrypt_preferred_challenge: Annotated[
+        Optional[LETSENCRYPT_PREFERRED_CHALLENGE],
+        typer.Option(help="Select preferred letsencrypt challenge.", show_default=False),
+    ] = None,
+    letsencrypt_email: Annotated[
+        Optional[str],
+        typer.Option(help="Specify email for letsencrypt", show_default=False),
+    ] = None,
     environment: Annotated[
         Optional[FMBenchEnvType],
-        typer.Option("--environment", help="Switch bench environment.", show_default=False),
+        typer.Option("--environment", "--env", help="Switch bench environment.", show_default=False),
     ] = None,
     developer_mode: Annotated[
         Optional[EnableDisableOptionsEnum],
@@ -530,20 +564,38 @@ def update(
         new_ssl_certificate = SSLCertificate(domain=benchname, ssl_type=SUPPORTED_SSL_TYPES.none)
 
         if ssl == SUPPORTED_SSL_TYPES.le:
-            if fm_config_manager.le_email == '[email protected]':
-                email = richprint.prompt_ask(prompt='Please enter [bold][green]email[/bold][/green] for Let\'s Encrypt')
+            if not letsencrypt_preferred_challenge:
+                if fm_config_manager.letsencrypt.exists:
+                    if letsencrypt_preferred_challenge is None:
+                        letsencrypt_preferred_challenge = LETSENCRYPT_PREFERRED_CHALLENGE.dns01
+
+                if not letsencrypt_preferred_challenge:
+                    letsencrypt_preferred_challenge = LETSENCRYPT_PREFERRED_CHALLENGE.http01
+
+            if fm_config_manager.letsencrypt.email == '[email protected]' or fm_config_manager.letsencrypt.email is None:
+                if not letsencrypt_email:
+                    richprint.stop()
+                    raise typer.BadParameter(
+                        "No email provided, required by certbot.", param_hint='--letsencrypt-email'
+                    )
+                else:
+                    email = letsencrypt_email
+
                 validate_email(email, check_deliverability=False)
-                fm_config_manager.le_email = email
-                fm_config_manager.export_to_toml()
+            else:
                 richprint.print(
-                    "Let's Encrypt email saved to configuration. It will be used automatically from now on."
+                    "Defaulting to Let's Encrypt email from [blue]fm_config.toml[/blue] since [blue]'--letsencrypt-email'[/blue] is not given."
                 )
-            else:
-                richprint.print("Using Let's Encrypt email from configuration.")
-                email = fm_config_manager.le_email
-                fm_config_manager.export_to_toml()
-
-            new_ssl_certificate = LetsencryptSSLCertificate(domain=benchname, ssl_type=ssl, email=email)
+                email = fm_config_manager.letsencrypt.email
+
+            new_ssl_certificate = LetsencryptSSLCertificate(
+                domain=benchname,
+                ssl_type=ssl,
+                email=email,
+                preferred_challenge=letsencrypt_preferred_challenge,
+                api_key=fm_config_manager.letsencrypt.api_key,
+                api_token=fm_config_manager.letsencrypt.api_token,
+            )
 
         richprint.print("Updating Certificate.")
         bench.update_certificate(new_ssl_certificate)

frappe_manager/metadata_manager.py

@@ -1,21 +1,54 @@
+from typing import Optional
 from pathlib import Path
-from pydantic import BaseModel, EmailStr
+from pydantic import BaseModel, EmailStr, Field
 import tomlkit
 from frappe_manager.migration_manager.version import Version
 from frappe_manager import CLI_FM_CONFIG_PATH
 from frappe_manager.utils.helpers import get_current_fm_version
 
 
+class FMLetsencryptConfig(BaseModel):
+    email: Optional[EmailStr] = Field(None, description="Email used by certbot.")
+    api_token: Optional[str] = Field(None, description="Cloudflare API token used by Certbot.")
+    api_key: Optional[str] = Field(None, description="Cloudflare Global API Key used by Certbot.")
+
+    @property
+    def exists(self):
+        if self.api_token or self.api_key:
+            return True
+
+        return False
+
+    def get_toml_doc(self):
+        model_dict = self.model_dump(exclude_none=True)
+        toml_doc = tomlkit.document()
+
+        for key, value in model_dict.items():
+            if isinstance(value, Path):
+                toml_doc[key] = str(value.absolute())
+            else:
+                toml_doc[key] = value
+        return toml_doc
+
+    @classmethod
+    def import_from_toml_doc(cls, toml_doc):
+        config_object = cls(**toml_doc)
+        return config_object
+
+
 class FMConfigManager(BaseModel):
     root_path: Path
     version: Version
-    le_email: EmailStr
+    letsencrypt: FMLetsencryptConfig = Field(default=FMLetsencryptConfig())
 
     def export_to_toml(self, path: Path = CLI_FM_CONFIG_PATH) -> bool:
         exclude = {'root_path'}
 
-        if self.le_email == '[email protected]':
-            exclude.add('le_email')
+        if not self.letsencrypt.email and not self.letsencrypt.api_key and not self.letsencrypt.api_token:
+            exclude.add('letsencrypt')
+        else:
+            if self.letsencrypt.email == '[email protected]':
+                exclude.add('letsencrypt')
 
         if self.version < Version('0.13.0'):
             path = CLI_FM_CONFIG_PATH.parent / '.fm.toml'
@@ -46,7 +79,7 @@ def import_from_toml(cls, path: Path = CLI_FM_CONFIG_PATH) -> "FMConfigManager":
         old_config_path = path.parent / '.fm.toml'
 
         input_data['version'] = Version('0.8.3')
-        input_data['le_email'] = '[email protected]'
+        input_data['letsencrypt'] = FMLetsencryptConfig(email=None, api_key=None, api_token=None)
         input_data['root_path'] = str(path)
 
         if old_config_path.exists():
@@ -55,7 +88,9 @@ def import_from_toml(cls, path: Path = CLI_FM_CONFIG_PATH) -> "FMConfigManager":
         elif path.exists():
             data = tomlkit.parse(path.read_text())
             input_data['version'] = Version(data.get('version', get_current_fm_version()))
-            input_data['le_email'] = data.get('le_email', '[email protected]')
+            input_data['letsencrypt'] = FMLetsencryptConfig(
+                **data.get('letsencrypt', {'email': None, 'api_key': None, 'api_token': None})
+            )
 
         fm_config_instance = cls(**input_data)
         return fm_config_instance

frappe_manager/migration_manager/migrations/migrate_0_13_0.py

@@ -2,6 +2,7 @@
 import os
 import copy
 from pathlib import Path
+import tomlkit
 from frappe_manager.compose_manager import DockerVolumeMount
 from frappe_manager.compose_manager.ComposeFile import ComposeFile
 from frappe_manager.migration_manager.migration_base import MigrationBase
@@ -14,9 +15,6 @@
 from frappe_manager.display_manager.DisplayManager import richprint
 from frappe_manager.migration_manager.version import Version
 from frappe_manager import CLI_DIR, CLI_SERVICES_DIRECTORY
-from frappe_manager.site_manager.bench_config import BenchConfig, FMBenchEnvType
-from frappe_manager.ssl_manager import SUPPORTED_SSL_TYPES
-from frappe_manager.ssl_manager.certificate import SSLCertificate
 from frappe_manager.utils.helpers import get_container_name_prefix
 from frappe_manager.docker_wrapper.DockerClient import DockerClient
 from frappe_manager.migration_manager.backup_manager import BackupManager
@@ -80,6 +78,7 @@ def migrate_services(self):
         # rename main config
         fm_config_path = CLI_DIR / 'fm_config.toml'
         old_fm_config_path = CLI_DIR / '.fm.toml'
+
         if old_fm_config_path.exists():
             old_fm_config_path.rename(fm_config_path)
 
@@ -133,43 +132,25 @@ def migrate_bench_compose(self, bench: MigrationBench):
         # create bench config
         frappe = envs.get('frappe', {})
 
-        userid = frappe.get('USERID', os.getuid())
-        usergroup = frappe.get('USERGROUP', os.getgid())
-
         apps_list = frappe.get('APPS_LIST', None)
 
         if apps_list:
             apps_list = apps_list.split(',')
         else:
             apps_list = []
 
-        frappe_branch = frappe.get('FRAPPE_BRANCH', 'version-15')
         developer_mode = frappe.get('DEVELOPER_MODE', True)
-        admin_pass = frappe.get('ADMIN_PASS', 'admin')
         name = frappe.get('SITENAME', bench.name)
-        mariadb_host = frappe.get('MARIADB_HOST', 'global-db')
-        mariadb_root_pass = frappe.get('MARIADB_ROOT_PASS', '/run/secrets/db_root_password')
-        environment_type = frappe.get('ENVIRONMENT', FMBenchEnvType.dev)
-        ssl_certificate = SSLCertificate(domain=bench.name, ssl_type=SUPPORTED_SSL_TYPES.none)
-
-        # TODO Handle admin tools compose change
-        bench_config = BenchConfig(
-            name=name,
-            userid=userid,
-            usergroup=usergroup,
-            apps_list=apps_list,
-            frappe_branch=frappe_branch,
-            developer_mode=developer_mode,
-            admin_tools=True,
-            admin_pass=admin_pass,
-            mariadb_host=mariadb_host,
-            mariadb_root_pass=mariadb_root_pass,
-            environment_type=environment_type,
-            root_path=bench_config_path,
-            ssl=ssl_certificate,
-        )
 
-        bench_config.export_to_toml(bench_config_path)
+        bench_config = tomlkit.document()
+
+        bench_config['name'] = name
+        bench_config['developer_mode'] = developer_mode
+        bench_config['admin_tools'] = True
+        bench_config['environment_type'] = 'dev'
+
+        with open(bench_config_path, 'w') as f:
+            f.write(tomlkit.dumps(bench_config))
 
         images_info = bench.compose_project.compose_file_manager.get_all_images()
 

frappe_manager/migration_manager/migrations/migrate_0_14_0.py

@@ -0,0 +1,36 @@
+from pathlib import Path
+import tomlkit
+from frappe_manager.migration_manager.migration_base import MigrationBase
+from frappe_manager.migration_manager.migration_helpers import MigrationBenches, MigrationServicesManager
+from frappe_manager.migration_manager.version import Version
+from frappe_manager.migration_manager.backup_manager import BackupManager
+from frappe_manager.display_manager.DisplayManager import richprint
+
+
+class MigrationV0140(MigrationBase):
+    version = Version("0.14.0")
+
+    def init(self):
+        self.cli_dir: Path = Path.home() / 'frappe'
+        self.benches_dir = self.cli_dir / "sites"
+        self.backup_manager = BackupManager(str(self.version), self.benches_dir)
+        self.benches_manager = MigrationBenches(self.benches_dir)
+        self.services_manager: MigrationServicesManager = MigrationServicesManager(
+            services_path=self.cli_dir / 'services'
+        )
+
+    def migrate_services(self):
+        richprint.change_head("Migrating [blue]fm_config.toml[/blue] changes")
+        config_path = self.cli_dir / 'fm_config.toml'
+
+        if config_path.exists():
+            data = tomlkit.parse(config_path.read_text())
+            email = data.get('le_email', None)
+
+            if email:
+                data['letsencrypt'] = {'email': email}
+                del data['le_email']
+
+            with open(config_path, 'w') as f:
+                f.write(tomlkit.dumps(data))
+        richprint.print("Migrated [blue]fm_config.toml[/blue]")

frappe_manager/site_manager/bench_config.py

@@ -5,7 +5,8 @@
 from typing import Any, List, Optional
 from pydantic import BaseModel, Field, model_validator, validator
 from frappe_manager import STABLE_APP_BRANCH_MAPPING_LIST
-from frappe_manager.ssl_manager import SUPPORTED_SSL_TYPES
+from frappe_manager.metadata_manager import FMConfigManager, FMLetsencryptConfig
+from frappe_manager.ssl_manager import LETSENCRYPT_PREFERRED_CHALLENGE, SUPPORTED_SSL_TYPES
 from frappe_manager.ssl_manager.certificate import SSLCertificate
 from frappe_manager.ssl_manager.letsencrypt_certificate import LetsencryptSSLCertificate
 from frappe_manager.utils.helpers import get_container_name_prefix
@@ -112,7 +113,27 @@ def import_from_toml(cls, path: Path) -> "BenchConfig":
             ssl_type = ssl_data.get('ssl_type', SUPPORTED_SSL_TYPES.none)
             if ssl_type == SUPPORTED_SSL_TYPES.le:
                 email = ssl_data.get('email', None)
-                ssl_instance = LetsencryptSSLCertificate(domain=domain, ssl_type=ssl_type, email=email)
+
+                fm_config_manager = FMConfigManager.import_from_toml()
+
+                pref_challenge_data = data.get("preferred_challenge", None)
+
+                if not pref_challenge_data:
+                    if fm_config_manager.letsencrypt.exists:
+                        preferred_challenge = LETSENCRYPT_PREFERRED_CHALLENGE.dns01
+                    else:
+                        preferred_challenge = LETSENCRYPT_PREFERRED_CHALLENGE.http01
+                else:
+                    preferred_challenge = pref_challenge_data
+
+                ssl_instance = LetsencryptSSLCertificate(
+                    domain=domain,
+                    ssl_type=ssl_type,
+                    email=email,
+                    preferred_challenge=preferred_challenge,
+                    api_key=fm_config_manager.letsencrypt.api_key,
+                    api_token=fm_config_manager.letsencrypt.api_token,
+                )
             else:
                 ssl_instance = SSLCertificate(domain=domain, ssl_type=SUPPORTED_SSL_TYPES.none)
         else:

frappe_manager/ssl_manager/__init__.py

@@ -4,3 +4,8 @@
 class SUPPORTED_SSL_TYPES(str, Enum):
     le = 'letsencrypt'
     none = 'disable'
+
+
+class LETSENCRYPT_PREFERRED_CHALLENGE(str, Enum):
+    dns01 = 'dns01'
+    http01 = 'http01'

frappe_manager/ssl_manager/certificate.py

@@ -14,16 +14,3 @@ class SSLCertificate(BaseModel):
     @property
     def has_wildcard(self) -> bool:
         return any(is_wildcard_fqdn(domain) for domain in self.alias_domains)
-
-
-class LetsencryptSSLCertificate(BaseModel):
-    domain: str
-    email: str
-    ssl_type: SUPPORTED_SSL_TYPES
-    hsts: str = 'off'
-    alias_domains: List[str] = []
-    toml_exclude: Optional[set] = {'domain', 'alias_domains', 'toml_exclude'}
-
-    @property
-    def has_wildcard(self) -> bool:
-        return any(is_wildcard_fqdn(domain) for domain in self.alias_domains)