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_path = install_path / "mkcert"
31
32 if force_reinstall and binary_path.exists():
33 click.secho("Removing existing mkcert binary...", bold=True)
34 binary_path.unlink()
35
36 if not binary_path.exists():
37 self._download_mkcert(binary_path)
38
39 self.mkcert_bin = str(binary_path)
40
41 # Run install if CA files don't exist, or if force reinstall
42 if force_reinstall or not self._ca_files_exist():
43 self.install_ca()
44
45 def _download_mkcert(self, dest: Path) -> None:
46 """Download the mkcert binary."""
47 system = platform.system()
48 machine = platform.machine().lower()
49
50 # Map platform.machine() to mkcert's expected architecture strings
51 arch_map = {
52 "x86_64": "amd64",
53 "amd64": "amd64",
54 "arm64": "arm64",
55 "aarch64": "arm64",
56 }
57 arch = arch_map.get(machine, "amd64")
58
59 os_map = {
60 "Darwin": "darwin",
61 "Linux": "linux",
62 "Windows": "windows",
63 }
64 os_name = os_map.get(system)
65 if not os_name:
66 click.secho(f"Unsupported OS: {system}", fg="red")
67 sys.exit(1)
68
69 mkcert_url = f"https://dl.filippo.io/mkcert/latest?for={os_name}/{arch}"
70 click.secho(f"Downloading mkcert from {mkcert_url}...", bold=True)
71
72 # Download to a temp file first, then atomically move it into place
73 # (the cache is machine-shared, and an interrupted download must not
74 # leave a partial binary behind at the final path).
75 tmp_path = dest.parent / f".download-{os.getpid()}"
76 try:
77 urllib.request.urlretrieve(mkcert_url, tmp_path)
78 tmp_path.chmod(0o755)
79 os.replace(tmp_path, dest)
80 finally:
81 tmp_path.unlink(missing_ok=True)
82
83 def _get_ca_root(self) -> Path | None:
84 """Get the mkcert CAROOT directory."""
85 if not self.mkcert_bin:
86 return None
87 result = subprocess.run(
88 [self.mkcert_bin, "-CAROOT"],
89 capture_output=True,
90 text=True,
91 check=False,
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"], check=False)
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 (
132 not force_regenerate
133 and cert_path.exists()
134 and key_path.exists()
135 and timestamp_path.exists()
136 ):
137 last_updated = timestamp_path.stat().st_mtime
138 if time.time() - last_updated < update_interval:
139 return cert_path, key_path
140
141 storage_path.mkdir(parents=True, exist_ok=True)
142
143 if not self.mkcert_bin:
144 raise RuntimeError("mkcert is not set up. Call setup_mkcert first.")
145
146 click.secho(f"Generating SSL certificates for {domain}...", bold=True)
147 subprocess.run(
148 [
149 self.mkcert_bin,
150 "-cert-file",
151 str(cert_path),
152 "-key-file",
153 str(key_path),
154 domain,
155 ],
156 check=True,
157 )
158
159 # Update the timestamp file to the current time
160 with open(timestamp_path, "w") as f:
161 f.write(str(time.time()))
162
163 return cert_path, key_path