diff --git a/functions/revoke.py b/functions/revoke.py index f5003e1..44c2800 100644 --- a/functions/revoke.py +++ b/functions/revoke.py @@ -4,11 +4,120 @@ Let's Encrypt certificate delete/revoke gates. from __future__ import annotations +from functions.common import log +from functions.common import replace_exactly_once + + +DELETE_FUNCTION_ANCHOR = '''\tdelete: async (access, data) => { +\t\tawait access.can("certificates:delete", data.id); +\t\tconst row = await internalCertificate.get(access, { id: data.id }); + +\t\tif (!row?.id) { +\t\t\tthrow new error.ItemNotFoundError(data.id); +\t\t} +''' + +DELETE_GATE_FRAGMENT = '''\tdelete: async (access, data) => { +\t\tawait access.can("certificates:delete", data.id); +\t\tconst row = await internalCertificate.get(access, { id: data.id }); + +\t\tif (!row?.id) { +\t\t\tthrow new error.ItemNotFoundError(data.id); +\t\t} + +\t\tif (row.provider === "letsencrypt") { +\t\t\trequireAcmeAuthority(); +\t\t} +''' + +REVOKE_FUNCTION_ANCHOR = '''\trevokeLetsEncryptSsl: async (certificate, throwErrors) => { +''' + +REVOKE_GATE_FRAGMENT = '''\trevokeLetsEncryptSsl: async (certificate, throwErrors) => { +\t\trequireAcmeAuthority(); +''' + def apply(source: str) -> str: """ - Install both: - - delete() pre-database-mutation authority gate - - revokeLetsEncryptSsl() execution authority gate + Install both Let's Encrypt revoke protections: + + 1. delete() must require authority before mutating the DB row. + 2. revokeLetsEncryptSsl() requires authority again immediately before + the Certbot revocation path. """ - raise NotImplementedError("revoke gate not implemented yet") + + if DELETE_GATE_FRAGMENT in source: + log("delete_gate=already-present") + else: + source = replace_exactly_once( + source, + DELETE_FUNCTION_ANCHOR, + DELETE_GATE_FRAGMENT, + description="delete gate", + ) + log("delete_gate=added") + + if REVOKE_GATE_FRAGMENT in source: + log("revoke_gate=already-present") + else: + source = replace_exactly_once( + source, + REVOKE_FUNCTION_ANCHOR, + REVOKE_GATE_FRAGMENT, + description="revoke gate", + ) + log("revoke_gate=added") + + return source + + +def verify(source: str) -> None: + """ + Verify both delete and Certbot revoke gates. + + Also verify that the delete gate appears before the database mutation. + """ + + if DELETE_GATE_FRAGMENT not in source: + raise RuntimeError( + "revoke gate verification: delete() authority check not found" + ) + + if REVOKE_GATE_FRAGMENT not in source: + raise RuntimeError( + "revoke gate verification: revokeLetsEncryptSsl() " + "authority check not found" + ) + + delete_gate_index = source.index(DELETE_GATE_FRAGMENT) + + delete_db_anchor = '''certificateModel.query().where("id", row.id).patch({ +\t\t\tis_deleted: 1, +\t\t});''' + + delete_db_index = source.find( + delete_db_anchor, + delete_gate_index, + ) + + if delete_db_index == -1: + raise RuntimeError( + "revoke gate verification: certificate delete DB mutation " + "not found after authority gate" + ) + + revoke_index = source.index(REVOKE_GATE_FRAGMENT) + + certbot_revoke_index = source.find( + '"revoke"', + revoke_index, + ) + + if certbot_revoke_index == -1: + raise RuntimeError( + "revoke gate verification: Certbot revoke operation " + "not found after authority gate" + ) + + log("revoke_gate_verification=passed")