1import os
2import platform
3import shutil
4import subprocess
5import sys
6import time
7import urllib.request
8from pathlib import Path
9
10import click
11
12from plain.runtime import PLAIN_CACHE_PATH
13
14
15class MkcertManager:
16 def __init__(self) -> None:
17 self.mkcert_bin: str | None = None
18
19 def setup_mkcert(self, *, force_reinstall: bool = False) -> None:
20 """Set up mkcert by checking if it's installed or downloading the binary and installing the local CA."""
21 if mkcert_path := shutil.which("mkcert"):
22 self.mkcert_bin = mkcert_path
23 # Run install if CA files don't exist, or if force reinstall
24 if force_reinstall or not self._ca_files_exist():
25 self.install_ca()
26 return
27
28 # mkcert not found system-wide, download to the machine-level cache
29 install_path = PLAIN_CACHE_PATH / "mkcert"
30 install_path.mkdir(parents=True, exist_ok=True)
31 binary_path = install_path / "mkcert"
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 )
93 if result.returncode == 0:
94 return Path(result.stdout.strip())
95 return None
96
97 def _ca_files_exist(self) -> bool:
98 """Check if the CA root files exist."""
99 ca_root = self._get_ca_root()
100 if not ca_root:
101 return False
102 return (ca_root / "rootCA.pem").exists() and (
103 ca_root / "rootCA-key.pem"
104 ).exists()
105
106 def install_ca(self) -> None:
107 """Install the mkcert CA into the system trust store.
108
109 Running `mkcert -install` is idempotent - if already installed,
110 it just prints a message without prompting for a password.
111 """
112 if not self.mkcert_bin:
113 return
114
115 # Don't capture output so user can see messages and respond to password prompts
116 result = subprocess.run([self.mkcert_bin, "-install"])
117
118 if result.returncode != 0:
119 click.secho("Failed to install mkcert CA", fg="red")
120 raise SystemExit(1)
121
122 def generate_certs(
123 self, domain: str, storage_path: Path, *, force_regenerate: bool = False
124 ) -> tuple[Path, Path]:
125 cert_path = storage_path / f"{domain}-cert.pem"
126 key_path = storage_path / f"{domain}-key.pem"
127 timestamp_path = storage_path / f"{domain}.timestamp"
128 update_interval = 60 * 24 * 3600 # 60 days in seconds
129
130 # Check if the certs exist and if the timestamp is recent enough
131 if not force_regenerate:
132 if cert_path.exists() and key_path.exists() and timestamp_path.exists():
133 last_updated = timestamp_path.stat().st_mtime
134 if time.time() - last_updated < update_interval:
135 return cert_path, key_path
136
137 storage_path.mkdir(parents=True, exist_ok=True)
138
139 if not self.mkcert_bin:
140 raise RuntimeError("mkcert is not set up. Call setup_mkcert first.")
141
142 click.secho(f"Generating SSL certificates for {domain}...", bold=True)
143 subprocess.run(
144 [
145 self.mkcert_bin,
146 "-cert-file",
147 str(cert_path),
148 "-key-file",
149 str(key_path),
150 domain,
151 ],
152 check=True,
153 )
154
155 # Update the timestamp file to the current time
156 with open(timestamp_path, "w") as f:
157 f.write(str(time.time()))
158
159 return cert_path, key_path