From a57c47d89bb73d7967a6cfbf26c6db2d41e1ca8d Mon Sep 17 00:00:00 2001 From: Sergey Matveev Date: Fri, 22 Jan 2021 13:19:55 +0300 Subject: [PATCH] Ability to issue child certificate --- pygost/asn1schemas/cert-selfsigned-example.py | 51 +++++++++++++++---- 1 file changed, 42 insertions(+), 9 deletions(-) diff --git a/pygost/asn1schemas/cert-selfsigned-example.py b/pygost/asn1schemas/cert-selfsigned-example.py index 20e873b..94b4b34 100755 --- a/pygost/asn1schemas/cert-selfsigned-example.py +++ b/pygost/asn1schemas/cert-selfsigned-example.py @@ -3,6 +3,7 @@ """ from argparse import ArgumentParser +from base64 import standard_b64decode from base64 import standard_b64encode from datetime import datetime from datetime import timedelta @@ -81,8 +82,12 @@ parser.add_argument( required=True, help="Signing algorithm: {256[ABCD],512[ABC]}", ) +parser.add_argument( + "--issue-with", + help="Path to PEM with CA to issue the child", +) args = parser.parse_args() -ai = { +AIs = { "256A": { "publicKeyParamSet": id_tc26_gost3410_2012_256_paramSetA, "key_algorithm": id_tc26_gost3410_2012_256, @@ -139,7 +144,34 @@ ai = { "sign_algorithm": id_tc26_signwithdigest_gost3410_2012_512, "hasher": GOST34112012512, }, -}[args.ai] +} +ai = AIs[args.ai] + +ca_prv = None +ca_subj = None +ca_ai = None +if args.issue_with is not None: + with open(args.issue_with, "rb") as fd: + lines = fd.read().decode("ascii").split("-----") + idx = lines.index("BEGIN PRIVATE KEY") + if idx == -1: + raise ValueError("PEM has no PRIVATE KEY") + prv_raw = standard_b64decode(lines[idx + 1]) + idx = lines.index("BEGIN CERTIFICATE") + if idx == -1: + raise ValueError("PEM has no CERTIFICATE") + cert_raw = standard_b64decode(lines[idx + 1]) + pki = PrivateKeyInfo().decod(prv_raw) + ca_prv = prv_unmarshal(bytes(OctetString().decod(bytes(pki["privateKey"])))) + tbs = Certificate().decod(cert_raw)["tbsCertificate"] + ca_subj = tbs["subject"] + curve_oid = GostR34102012PublicKeyParameters().decod(bytes( + tbs["subjectPublicKeyInfo"]["algorithm"]["parameters"] + ))["publicKeyParamSet"] + ca_ai = next(iter([ + params for params in AIs.values() + if params["publicKeyParamSet"] == curve_oid + ])) def pem(obj): @@ -176,11 +208,12 @@ subj = Name(("rdnSequence", RDNSequence([ not_before = datetime.utcnow() not_after = not_before + timedelta(days=365) ai_sign = AlgorithmIdentifier(( - ("algorithm", ai["sign_algorithm"],), + ("algorithm", (ai if ca_ai is None else ca_ai)["sign_algorithm"]), )) exts = [ Extension(( ("extnID", id_ce_subjectKeyIdentifier), + ("extnValue", OctetString( SubjectKeyIdentifier(GOST34112012256(pub_raw).digest()[:20]).encode() )), @@ -203,7 +236,7 @@ tbs = TBSCertificate(( ("version", Version("v3")), ("serialNumber", CertificateSerialNumber(12345)), ("signature", ai_sign), - ("issuer", subj), + ("issuer", subj if ca_ai is None else ca_subj), ("validity", Validity(( ("notBefore", Time(("utcTime", UTCTime(not_before)))), ("notAfter", Time(("utcTime", UTCTime(not_after)))), @@ -221,11 +254,11 @@ tbs = TBSCertificate(( cert = Certificate(( ("tbsCertificate", tbs), ("signatureAlgorithm", ai_sign), - ("signatureValue", BitString(sign( - curve, - prv, - ai["hasher"](tbs.encode()).digest()[::-1], - ))), + ("signatureValue", BitString( + sign(curve, prv, ai["hasher"](tbs.encode()).digest()[::-1]) + if ca_ai is None else + sign(ca_ai["curve"], ca_prv, ca_ai["hasher"](tbs.encode()).digest()[::-1]) + )), )) print("-----BEGIN CERTIFICATE-----") print(pem(cert)) -- 2.44.0