Add ACME certificate request gate mutation
This commit is contained in:
+56
-2
@@ -4,9 +4,63 @@ Let's Encrypt certificate request/create gate.
|
|||||||
|
|
||||||
from __future__ import annotations
|
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:
|
def apply(source: str) -> str:
|
||||||
"""
|
"""
|
||||||
Gate NPM create() for provider == letsencrypt before DB insertion.
|
Gate Let's Encrypt certificate creation before database insertion.
|
||||||
|
|
||||||
|
Custom certificate creation remains unaffected.
|
||||||
"""
|
"""
|
||||||
raise NotImplementedError("request gate not implemented yet")
|
|
||||||
|
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")
|
||||||
|
|||||||
Reference in New Issue
Block a user