1import os
2import platform
3import shutil
4import subprocess
5import sys
6import time
7import urllib.request
8from pathlib import Path
9
10import click
11from plain.runtime import PLAIN_CACHE_PATH
12
13
14class MkcertManager:
15 def __init__(self) -> None:
16 self.mkcert_bin: str | None = None
17
18 def setup_mkcert(self, *, force_reinstall: bool = False) -> None:
19 """Set up mkcert by checking if it's installed or downloading the binary and installing the local CA."""
20 if mkcert_path := shutil.which("mkcert"):
21 self.mkcert_bin = mkcert_path
22 # Run install if CA files don't exist, or if force reinstall
23 if force_reinstall or not self._ca_files_exist():
24 self.install_ca()
25 return
26
27 # mkcert not found system-wide, download to the machine-level cache
28 install_path = PLAIN_CACHE_PATH / "mkcert"
29 install_path.mkdir(parents=True, exist_ok=True)
30 binary_name = "mkcert.exe" if platform.system() == "Windows" else "mkcert"
31 binary_path = install_path / binary_name
32
33 if force_reinstall and binary_path.exists():
34 click.secho("Removing existing mkcert binary...", bold=True)
35 binary_path.unlink()
36
37 if not binary_path.exists():
38 self._download_mkcert(binary_path)
39
40 self.mkcert_bin = str(binary_path)
41
42 # Run install if CA files don't exist, or if force reinstall
43 if force_reinstall or not self._ca_files_exist():
44 self.install_ca()
45
46 def _download_mkcert(self, dest: Path) -> None:
47 """Download the mkcert binary."""
48 system = platform.system()
49 machine = platform.machine().lower()
50
51 # Map platform.machine() to mkcert's expected architecture strings
52 arch_map = {
53 "x86_64": "amd64",
54 "amd64": "amd64",
55 "arm64": "arm64",
56 "aarch64": "arm64",
57 }
58 arch = arch_map.get(machine, "amd64")
59
60 os_map = {
61 "Darwin": "darwin",
62 "Linux": "linux",
63 "Windows": "windows",
64 }
65 os_name = os_map.get(system)
66 if not os_name:
67 click.secho(f"Unsupported OS: {system}", fg="red")
68 sys.exit(1)
69
70 mkcert_url = f"https://dl.filippo.io/mkcert/latest?for={os_name}/{arch}"
71 click.secho(f"Downloading mkcert from {mkcert_url}...", bold=True)
72
73 # Download to a temp file first, then atomically move it into place
74 # (the cache is machine-shared, and an interrupted download must not
75 # leave a partial binary behind at the final path).
76 tmp_path = dest.parent / f".download-{os.getpid()}"
77 try:
78 urllib.request.urlretrieve(mkcert_url, tmp_path)
79 tmp_path.chmod(0o755)
80 os.replace(tmp_path, dest)
81 finally:
82 tmp_path.unlink(missing_ok=True)
83
84 def _get_ca_root(self) -> Path | None:
85 """Get the mkcert CAROOT directory."""
86 if not self.mkcert_bin:
87 return None
88 result = subprocess.run(
89 [self.mkcert_bin, "-CAROOT"],
90 capture_output=True,
91 text=True,
92 check=False,
93 )
94 if result.returncode == 0:
95 return Path(result.stdout.strip())
96 return None
97
98 def _ca_files_exist(self) -> bool:
99 """Check if the CA root files exist."""
100 ca_root = self._get_ca_root()
101 if not ca_root:
102 return False
103 return (ca_root / "rootCA.pem").exists() and (
104 ca_root / "rootCA-key.pem"
105 ).exists()
106
107 def install_ca(self) -> None:
108 """Install the mkcert CA into the system trust store.
109
110 Running `mkcert -install` is idempotent - if already installed,
111 it just prints a message without prompting for a password.
112 """
113 if not self.mkcert_bin:
114 return
115
116 # Don't capture output so user can see messages and respond to password prompts
117 result = subprocess.run([self.mkcert_bin, "-install"], check=False)
118
119 if result.returncode != 0:
120 click.secho("Failed to install mkcert CA", fg="red")
121 raise SystemExit(1)
122
123 def generate_certs(
124 self, domain: str, storage_path: Path, *, force_regenerate: bool = False
125 ) -> tuple[Path, Path]:
126 cert_path = storage_path / f"{domain}-cert.pem"
127 key_path = storage_path / f"{domain}-key.pem"
128 timestamp_path = storage_path / f"{domain}.timestamp"
129 update_interval = 60 * 24 * 3600 # 60 days in seconds
130
131 # Check if the certs exist and if the timestamp is recent enough
132 if (
133 not force_regenerate
134 and cert_path.exists()
135 and key_path.exists()
136 and timestamp_path.exists()
137 ):
138 last_updated = timestamp_path.stat().st_mtime
139 if time.time() - last_updated < update_interval:
140 return cert_path, key_path
141
142 storage_path.mkdir(parents=True, exist_ok=True)
143
144 if not self.mkcert_bin:
145 raise RuntimeError("mkcert is not set up. Call setup_mkcert first.")
146
147 click.secho(f"Generating SSL certificates for {domain}...", bold=True)
148 subprocess.run(
149 [
150 self.mkcert_bin,
151 "-cert-file",
152 str(cert_path),
153 "-key-file",
154 str(key_path),
155 domain,
156 ],
157 check=True,
158 )
159
160 # Update the timestamp file to the current time
161 with open(timestamp_path, "w") as f:
162 f.write(str(time.time()))
163
164 return cert_path, key_path