67 lines
1.6 KiB
Python
67 lines
1.6 KiB
Python
"""
|
|
Let's Encrypt certificate request/create gate.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from functions.common import log
|
|
from functions.common import replace_exactly_once
|
|
|
|
|
|
REQUEST_FUNCTION_ANCHOR = '''\tcreate: async (access, data) => {
|
|
\t\tawait access.can("certificates:create", data);
|
|
'''
|
|
|
|
REQUEST_GATE_FRAGMENT = '''\tcreate: async (access, data) => {
|
|
\t\tawait access.can("certificates:create", data);
|
|
|
|
\t\tif (data.provider === "letsencrypt") {
|
|
\t\t\trequireAcmeAuthority();
|
|
\t\t}
|
|
'''
|
|
|
|
|
|
def apply(source: str) -> str:
|
|
"""
|
|
Gate Let's Encrypt certificate creation before database insertion.
|
|
|
|
Custom certificate creation remains unaffected.
|
|
"""
|
|
|
|
if REQUEST_GATE_FRAGMENT in source:
|
|
log("request_gate=already-present")
|
|
return source
|
|
|
|
source = replace_exactly_once(
|
|
source,
|
|
REQUEST_FUNCTION_ANCHOR,
|
|
REQUEST_GATE_FRAGMENT,
|
|
description="request gate",
|
|
)
|
|
|
|
log("request_gate=added")
|
|
return source
|
|
|
|
|
|
def verify(source: str) -> None:
|
|
"""
|
|
Verify that create() contains the Let's Encrypt authority gate.
|
|
"""
|
|
|
|
if REQUEST_GATE_FRAGMENT not in source:
|
|
raise RuntimeError(
|
|
"request gate verification: authority check not found in create()"
|
|
)
|
|
|
|
gate_index = source.index(REQUEST_GATE_FRAGMENT)
|
|
|
|
db_anchor = "certificateModel.query().insertAndFetch(data)"
|
|
db_index = source.find(db_anchor, gate_index)
|
|
|
|
if db_index == -1:
|
|
raise RuntimeError(
|
|
"request gate verification: certificate DB insert not found after gate"
|
|
)
|
|
|
|
log("request_gate_verification=passed")
|