]> Cypherpunks.ru repositories - pygost.git/blobdiff - pygost/asn1schemas/cert-selfsigned-example.py
Ability to reuse the key for reissuing
[pygost.git] / pygost / asn1schemas / cert-selfsigned-example.py
index edce6969a92feef39d3d629fc575f26d40c64efa..ae1ee00281f1f5b53576ab1f065b6588832b4bea 100755 (executable)
@@ -8,6 +8,7 @@ from base64 import standard_b64encode
 from datetime import datetime
 from datetime import timedelta
 from os import urandom
+from sys import stdout
 from textwrap import fill
 
 from pyderasn import Any
@@ -20,6 +21,7 @@ from pyderasn import PrintableString
 from pyderasn import UTCTime
 
 from pygost.asn1schemas.oids import id_at_commonName
+from pygost.asn1schemas.oids import id_at_countryName
 from pygost.asn1schemas.oids import id_ce_authorityKeyIdentifier
 from pygost.asn1schemas.oids import id_ce_basicConstraints
 from pygost.asn1schemas.oids import id_ce_keyUsage
@@ -70,6 +72,7 @@ from pygost.gost3410 import public_key
 from pygost.gost3410 import sign
 from pygost.gost34112012256 import GOST34112012256
 from pygost.gost34112012512 import GOST34112012512
+from pygost.utils import bytes2long
 
 parser = ArgumentParser(description="Self-signed X.509 certificate creator")
 parser.add_argument(
@@ -82,6 +85,14 @@ parser.add_argument(
     required=True,
     help="Subject's CommonName",
 )
+parser.add_argument(
+    "--country",
+    help="Subject's Country",
+)
+parser.add_argument(
+    "--serial",
+    help="Serial number",
+)
 parser.add_argument(
     "--ai",
     required=True,
@@ -91,6 +102,23 @@ parser.add_argument(
     "--issue-with",
     help="Path to PEM with CA to issue the child",
 )
+parser.add_argument(
+    "--reuse-key",
+    help="Path to PEM with the key to reuse",
+)
+parser.add_argument(
+    "--out-key",
+    help="Path to PEM with the resulting key",
+)
+parser.add_argument(
+    "--only-key",
+    action="store_true",
+    help="Only generate the key",
+)
+parser.add_argument(
+    "--out-cert",
+    help="Path to PEM with the resulting certificate",
+)
 args = parser.parse_args()
 AIs = {
     "256A": {
@@ -180,38 +208,57 @@ if args.issue_with is not None:
         if params["publicKeyParamSet"] == curve_oid
     ]))
 
+key_params = GostR34102012PublicKeyParameters((
+    ("publicKeyParamSet", ai["publicKeyParamSet"]),
+))
+
 
 def pem(obj):
     return fill(standard_b64encode(obj.encode()).decode("ascii"), 64)
 
 
-key_params = GostR34102012PublicKeyParameters((
-    ("publicKeyParamSet", ai["publicKeyParamSet"]),
-))
-
-prv_raw = urandom(ai["prv_len"])
-print("-----BEGIN PRIVATE KEY-----")
-print(pem(PrivateKeyInfo((
-    ("version", Integer(0)),
-    ("privateKeyAlgorithm", PrivateKeyAlgorithmIdentifier((
-        ("algorithm", ai["key_algorithm"]),
-        ("parameters", Any(key_params)),
-    ))),
-    ("privateKey", PrivateKey(OctetString(prv_raw).encode())),
-))))
-print("-----END PRIVATE KEY-----")
+if args.reuse_key is not None:
+    with open(args.reuse_key, "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])
+    pki = PrivateKeyInfo().decod(prv_raw)
+    prv = prv_unmarshal(bytes(OctetString().decod(bytes(pki["privateKey"]))))
+else:
+    prv_raw = urandom(ai["prv_len"])
+    out = stdout if args.out_key is None else open(args.out_key, "w")
+    print("-----BEGIN PRIVATE KEY-----", file=out)
+    print(pem(PrivateKeyInfo((
+        ("version", Integer(0)),
+        ("privateKeyAlgorithm", PrivateKeyAlgorithmIdentifier((
+            ("algorithm", ai["key_algorithm"]),
+            ("parameters", Any(key_params)),
+        ))),
+        ("privateKey", PrivateKey(OctetString(prv_raw).encode())),
+    ))), file=out)
+    print("-----END PRIVATE KEY-----", file=out)
+    if args.only_key:
+        exit()
+    prv = prv_unmarshal(prv_raw)
 
-prv = prv_unmarshal(prv_raw)
 curve = ai["curve"]
 pub_raw = pub_marshal(public_key(curve, prv))
-subj = Name(("rdnSequence", RDNSequence([
-    RelativeDistinguishedName((
+rdn = [RelativeDistinguishedName((
+    AttributeTypeAndValue((
+        ("type", AttributeType(id_at_commonName)),
+        ("value", AttributeValue(PrintableString(args.cn))),
+    )),
+))]
+if args.country:
+    rdn.append(RelativeDistinguishedName((
         AttributeTypeAndValue((
-            ("type", AttributeType(id_at_commonName)),
-            ("value", AttributeValue(PrintableString(args.cn))),
+            ("type", AttributeType(id_at_countryName)),
+            ("value", AttributeValue(PrintableString(args.country))),
         )),
-    ))
-])))
+    )))
+subj = Name(("rdnSequence", RDNSequence(rdn)))
 not_before = datetime.utcnow()
 not_after = not_before + timedelta(days=365 * (10 if args.ca else 1))
 ai_sign = AlgorithmIdentifier((
@@ -220,18 +267,16 @@ ai_sign = AlgorithmIdentifier((
 exts = [
     Extension((
         ("extnID", id_ce_subjectKeyIdentifier),
-
         ("extnValue", OctetString(
             SubjectKeyIdentifier(GOST34112012256(pub_raw).digest()[:20]).encode()
         )),
     )),
     Extension((
-        ("extnID", id_ce_subjectAltName),
-        ("extnValue", OctetString(
-            SubjectAltName((
-                GeneralName(("dNSName", IA5String(args.cn))),
-            )).encode()
-        )),
+        ("extnID", id_ce_keyUsage),
+        ("critical", Boolean(True)),
+        ("extnValue", OctetString(KeyUsage(
+            ("keyCertSign" if args.ca else "digitalSignature",),
+        ).encode())),
     )),
 ]
 if args.ca:
@@ -242,10 +287,14 @@ if args.ca:
             ("cA", Boolean(True)),
         )).encode())),
     )))
+else:
     exts.append(Extension((
-        ("extnID", id_ce_keyUsage),
-        ("critical", Boolean(True)),
-        ("extnValue", OctetString(KeyUsage(("keyCertSign",)).encode())),
+        ("extnID", id_ce_subjectAltName),
+        ("extnValue", OctetString(
+            SubjectAltName((
+                GeneralName(("dNSName", IA5String(args.cn))),
+            )).encode()
+        )),
     )))
 if ca_ai is not None:
     caKeyId = [
@@ -260,9 +309,13 @@ if ca_ai is not None:
         )).encode())),
     )))
 
+serial = (
+    bytes2long(GOST34112012256(urandom(16)).digest()[:20])
+    if args.serial is None else int(args.serial)
+)
 tbs = TBSCertificate((
     ("version", Version("v3")),
-    ("serialNumber", CertificateSerialNumber(12345)),
+    ("serialNumber", CertificateSerialNumber(serial)),
     ("signature", ai_sign),
     ("issuer", subj if ca_ai is None else ca_subj),
     ("validity", Validity((
@@ -288,6 +341,7 @@ cert = Certificate((
         sign(ca_ai["curve"], ca_prv, ca_ai["hasher"](tbs.encode()).digest()[::-1])
     )),
 ))
-print("-----BEGIN CERTIFICATE-----")
-print(pem(cert))
-print("-----END CERTIFICATE-----")
+out = stdout if args.out_cert is None else open(args.out_cert, "w")
+print("-----BEGIN CERTIFICATE-----", file=out)
+print(pem(cert), file=out)
+print("-----END CERTIFICATE-----", file=out)