Add local production infra lab
This commit is contained in:
@@ -21,6 +21,7 @@ set -uo pipefail
|
||||
# CASAN_REVIEWERS_DIR (default governance/reviewers) — base dir for pubkey-file
|
||||
# CASAN_APPROVAL_JWT (optional RS256 IdP token)
|
||||
# CASAN_IDP_PUBLIC_KEY (default central-governance/idp-public.pem)
|
||||
# CASAN_IDP_JWKS_URL (optional OIDC JWKS endpoint; overrides public key)
|
||||
# Exit: 0 ok (prints "APPROVAL_OK role=<role>"), 3 deny (reason on stderr), 64 usage.
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
@@ -54,7 +55,11 @@ ROLES="$(awk -v a="$ACTION" '$1=="action" && $2==a {print $3; exit}' "$REVIEWERS
|
||||
|
||||
if [[ -n "${CASAN_APPROVAL_JWT:-}" ]]; then
|
||||
IDP_PUB="${CASAN_IDP_PUBLIC_KEY:-$GOV_DIR/idp-public.pem}"
|
||||
[[ -f "$IDP_PUB" ]] || deny "idp_pubkey_missing($IDP_PUB)"
|
||||
if [[ -n "${CASAN_IDP_JWKS_URL:-}" ]]; then
|
||||
IDP_PUB="jwks:${CASAN_IDP_JWKS_URL}"
|
||||
else
|
||||
[[ -f "$IDP_PUB" ]] || deny "idp_pubkey_missing($IDP_PUB)"
|
||||
fi
|
||||
JWT_OUT="$(
|
||||
python3 - "$CASAN_APPROVAL_JWT" "$IDP_PUB" "$APPROVER" "$ROLES" "$ACTION" "$ACTOR" "$INPUT_SHA" <<'PY'
|
||||
import base64
|
||||
@@ -64,6 +69,7 @@ import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.request
|
||||
|
||||
jwt, pub, approver, roles_csv, action, actor, input_sha = sys.argv[1:8]
|
||||
|
||||
@@ -101,13 +107,54 @@ if claims.get("input_sha256") != input_sha:
|
||||
|
||||
sig = b64u_decode(parts[2])
|
||||
signing_input = ".".join(parts[:2]).encode()
|
||||
|
||||
def der_len(n):
|
||||
if n < 128:
|
||||
return bytes([n])
|
||||
raw = n.to_bytes((n.bit_length() + 7) // 8, "big")
|
||||
return bytes([0x80 | len(raw)]) + raw
|
||||
|
||||
def der_tlv(tag, body):
|
||||
return bytes([tag]) + der_len(len(body)) + body
|
||||
|
||||
def der_int(n):
|
||||
raw = n.to_bytes((n.bit_length() + 7) // 8, "big") or b"\x00"
|
||||
if raw[0] & 0x80:
|
||||
raw = b"\x00" + raw
|
||||
return der_tlv(0x02, raw)
|
||||
|
||||
def jwk_to_pem(jwk):
|
||||
n = int.from_bytes(b64u_decode(jwk["n"]), "big")
|
||||
e = int.from_bytes(b64u_decode(jwk["e"]), "big")
|
||||
rsa_pub = der_tlv(0x30, der_int(n) + der_int(e))
|
||||
alg_id = der_tlv(
|
||||
0x30,
|
||||
der_tlv(0x06, b"\x2a\x86\x48\x86\xf7\x0d\x01\x01\x01") + der_tlv(0x05, b""),
|
||||
)
|
||||
spki = der_tlv(0x30, alg_id + der_tlv(0x03, b"\x00" + rsa_pub))
|
||||
b64 = base64.encodebytes(spki).decode().replace("\n", "")
|
||||
lines = [b64[i:i+64] for i in range(0, len(b64), 64)]
|
||||
return "-----BEGIN PUBLIC KEY-----\n" + "\n".join(lines) + "\n-----END PUBLIC KEY-----\n"
|
||||
|
||||
pub_file = pub
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
sig_path = os.path.join(td, "sig.bin")
|
||||
msg_path = os.path.join(td, "msg.txt")
|
||||
open(sig_path, "wb").write(sig)
|
||||
open(msg_path, "wb").write(signing_input)
|
||||
if pub.startswith("jwks:"):
|
||||
try:
|
||||
jwks = json.loads(urllib.request.urlopen(pub[len("jwks:"):], timeout=5).read().decode())
|
||||
except Exception:
|
||||
die("jwks_fetch_failed")
|
||||
kid = header.get("kid")
|
||||
keys = [k for k in jwks.get("keys", []) if k.get("kty") == "RSA" and (not kid or k.get("kid") == kid)]
|
||||
if not keys:
|
||||
die("jwks_key_not_found")
|
||||
pub_file = os.path.join(td, "idp-public.pem")
|
||||
open(pub_file, "w", encoding="utf-8").write(jwk_to_pem(keys[0]))
|
||||
rc = subprocess.run(
|
||||
["openssl", "dgst", "-sha256", "-verify", pub, "-signature", sig_path, msg_path],
|
||||
["openssl", "dgst", "-sha256", "-verify", pub_file, "-signature", sig_path, msg_path],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
).returncode
|
||||
|
||||
Reference in New Issue
Block a user