feat: app orchestrator, entry point, CLAUDE.md — implementation complete

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Vlad Doloman
2026-06-24 00:24:30 +03:00
parent ca6fe9f525
commit 1b2f4d2bdc
2 changed files with 243 additions and 2 deletions

View File

@@ -746,12 +746,196 @@ def show_main_screen(stdscr, certs: list[CertInfo]) -> ScreenResult:
return ScreenResult(action=Action.REVOKE, selected_cns=[cert.cn])
# (remaining sections added in later tasks)
# ============================================================
# === APP ===
# ============================================================
class CursesApp:
def run(self) -> None:
curses.wrapper(self._main)
def _main(self, stdscr) -> None:
curses.curs_set(0)
init_colors()
while True:
try:
certs = load_expiring_certs(
EASYRSA_PKI_DIR, DAYS_PAST, DAYS_AHEAD,
)
except FileNotFoundError as exc:
self._fatal(stdscr, f"Cannot read PKI index.txt:\n{exc}")
return
result = show_main_screen(stdscr, certs)
if result.action == Action.EXIT:
return
elif result.action == Action.REGEN_CRL:
self._regen_crl(stdscr)
elif result.action == Action.REVOKE:
self._do_revoke_and_crl(stdscr, result.selected_cns[0])
elif result.action == Action.NEW_CERT:
self._process_cert(stdscr, cn="", email="", is_renewal=False)
elif result.action == Action.RENEW_SELECTED:
for cn in result.selected_cns:
email_addr = get_email(EASYRSA_PKI_DIR, cn)
if not self._process_cert(stdscr, cn=cn, email=email_addr,
is_renewal=True):
break
# Loop back: reload cert list (it changed after renewal/revoke)
# ── Workflows ────────────────────────────────────────────────────────────
def _process_cert(
self, stdscr, cn: str, email: str, is_renewal: bool,
) -> bool:
"""Form → generate → deliver. Returns False if user cancelled."""
form = show_cert_form(stdscr, cn=cn, email=email, cn_readonly=is_renewal)
if form.cancelled:
return False
final_cn = form.cn
self._msg(stdscr, f"Generating certificate for {final_cn}")
try:
if is_renewal:
revoke_issued(EASYRSA_DIR, EASYRSA_PKI_DIR, final_cn, CA_PASSPHRASE)
build_client_full(EASYRSA_DIR, EASYRSA_PKI_DIR,
final_cn, form.password, CA_PASSPHRASE)
except EasyRSAError as exc:
self._error(stdscr, f"EasyRSA error:\n{exc}")
return True
if form.email:
save_email(EASYRSA_PKI_DIR, final_cn, form.email)
try:
ovpn_path = build_ovpn(final_cn, EASYRSA_PKI_DIR,
OVPN_TEMPLATE_PATH, VPN_CONFIGS_DIR, CONFIG_NAME)
except Exception as exc:
self._error(stdscr, f"Failed to build .ovpn:\n{exc}")
return True
try:
one_time_url = create_note(form.password, CRYPTGEON_URL)
except CryptgeonError as exc:
self._error(stdscr, f"Cryptgeon error:\n{exc}")
self._show_result(stdscr, final_cn, ovpn_path, form.password,
one_time_url="(Cryptgeon failed — see error above)")
return True
if form.email and show_confirm(
stdscr,
f"Send email to {form.email}?\n\n"
f"Attachment : {os.path.basename(ovpn_path)}\n"
f"Password URL: {one_time_url[:50]}",
):
try:
send_email(form.email, final_cn, one_time_url, ovpn_path,
MAIL_FROM, MAIL_SUBJECT, EMAIL_TEMPLATE_PATH, MAIL_BINARY)
self._msg(stdscr,
f"Email sent to {form.email}\nConfig: {ovpn_path}",
wait=True)
except RuntimeError as exc:
self._error(stdscr,
f"Email failed:\n{exc}\n\n"
f"URL : {one_time_url}\n"
f"Config: {ovpn_path}")
else:
self._show_result(stdscr, final_cn, ovpn_path, form.password, one_time_url)
return True
def _do_revoke_and_crl(self, stdscr, cn: str) -> None:
self._msg(stdscr, f"Revoking {cn}")
try:
revoke_issued(EASYRSA_DIR, EASYRSA_PKI_DIR, cn, CA_PASSPHRASE)
gen_crl(EASYRSA_DIR, EASYRSA_PKI_DIR, CA_PASSPHRASE)
copy_crl(EASYRSA_PKI_DIR, CRL_DEST_PATH)
except EasyRSAError as exc:
self._error(stdscr, f"Revoke failed:\n{exc}")
return
self._msg(stdscr, f"{cn} revoked and CRL updated → {CRL_DEST_PATH}", wait=True)
def _regen_crl(self, stdscr) -> None:
self._msg(stdscr, "Regenerating CRL…")
try:
gen_crl(EASYRSA_DIR, EASYRSA_PKI_DIR, CA_PASSPHRASE)
copy_crl(EASYRSA_PKI_DIR, CRL_DEST_PATH)
except EasyRSAError as exc:
self._error(stdscr, f"gen-crl failed:\n{exc}")
return
self._msg(stdscr, f"CRL regenerated → {CRL_DEST_PATH}", wait=True)
# ── Display helpers ───────────────────────────────────────────────────────
def _msg(self, stdscr, text: str, wait: bool = False) -> None:
stdscr.clear()
sh, sw = stdscr.getmaxyx()
lines = text.splitlines()
start_y = max(1, (sh - len(lines)) // 2)
for i, line in enumerate(lines):
try:
stdscr.addstr(start_y + i,
max(0, (sw - len(line)) // 2),
line[:sw], curses.color_pair(COLOR_NORMAL))
except curses.error:
pass
if wait:
try:
stdscr.addstr(sh - 2, 2, "Press any key to continue…",
curses.color_pair(COLOR_DISABLED))
except curses.error:
pass
stdscr.refresh()
if wait:
stdscr.getch()
def _error(self, stdscr, text: str) -> None:
lines = ["── ERROR ──", ""] + text.splitlines() + ["", "Press any key…"]
stdscr.clear()
sh, sw = stdscr.getmaxyx()
start_y = max(1, (sh - len(lines)) // 2)
for i, line in enumerate(lines):
attr = (curses.color_pair(COLOR_ERROR) | curses.A_BOLD
if i == 0 else curses.color_pair(COLOR_NORMAL))
try:
stdscr.addstr(start_y + i,
max(0, (sw - len(line)) // 2),
line[:sw], attr)
except curses.error:
pass
stdscr.refresh()
stdscr.getch()
def _fatal(self, stdscr, text: str) -> None:
self._error(stdscr, text)
def _show_result(
self, stdscr, cn: str, ovpn_path: str, password: str, one_time_url: str,
) -> None:
self._msg(
stdscr,
f"Certificate issued for: {cn}\n"
f"\n"
f"Config file:\n {ovpn_path}\n"
f"\n"
f"Password URL (one-time):\n {one_time_url}\n"
f"\n"
f"Password (shown once):\n {password}",
wait=True,
)
# ============================================================
# === ENTRY POINT ===
# ============================================================
def main() -> None:
pass # replaced in Task 11
CursesApp().run()
if __name__ == "__main__":