124 lines
3.2 KiB
Python
124 lines
3.2 KiB
Python
"""
|
|
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 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.
|
|
"""
|
|
|
|
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")
|