feat: Bring Go and Python implementations to feature parity
Go: - Read and write simple-ca.json shared with the Python implementation - Add make-crl and revoke-cert commands and record certificate history - Rename --aia-base-url to --ca-publish-base-url - Add CRL Distribution Point extension to issuing CA and leaf certificates - Take days and issuing_ca defaults from simple-ca.json - Include email SANs in server certificates - Reject IPv4 SANs with out-of-range octets - Record existing issuing CAs and rebuild the CA bundle - Print errors once Python: - Add make-cert --type server|user and email SANs - Reject IPv4 SANs with out-of-range octets - Skip make-ca when the root CA exists and no issuing CA is given - Build ca_bundle.pem from issuing CA subdirectories - Accept a bare filename in make-pfx and drop the root key requirement - Honour explicit --days values and reject values below 1 - Persist and increment CRL numbers in simple-ca.json Update README and add tests for the new behaviour.
This commit is contained in:
slawek committed
1 parent
27461a0fbe
commit
9c84e82672
5 files changed
+757
-170
No files matched your search
+95
-49
@@ -25,6 +25,7 @@
|
||||
|
||||
import argparse
|
||||
import datetime
|
||||
import ipaddress
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
@@ -103,7 +104,7 @@ def _rebuild_ca_bundle(ca_dir):
|
||||
if os.path.isfile(root):
|
||||
with open(root, "rb") as f:
|
||||
parts.append(f.read())
|
||||
for name in sorted(_config.get("subordinates", [])):
|
||||
for name in sorted(os.listdir(ca_dir)):
|
||||
sub_cert = os.path.join(ca_dir, name, "ca_cert.pem")
|
||||
if os.path.isfile(sub_cert):
|
||||
with open(sub_cert, "rb") as f:
|
||||
@@ -140,6 +141,7 @@ def _iso_to_asn1(iso: str) -> str:
|
||||
|
||||
_IP_RE = re.compile(r"^[0-9]{1,3}(\.[0-9]{1,3}){3}$")
|
||||
_DNS_RE = re.compile(r"^[a-z0-9-]+(\.[a-z0-9-]+)*$")
|
||||
_EMAIL_RE = re.compile(r"^[^@]+@[^@]+\.[^@]+$")
|
||||
|
||||
|
||||
def _is_ip(value):
|
||||
@@ -150,6 +152,10 @@ def _is_dns(value):
|
||||
return bool(_DNS_RE.match(value))
|
||||
|
||||
|
||||
def _is_email(value):
|
||||
return bool(_EMAIL_RE.match(value))
|
||||
|
||||
|
||||
def _pipe(cmd1, cmd2):
|
||||
p1 = subprocess.Popen(cmd1, stdout=subprocess.PIPE)
|
||||
p2 = subprocess.Popen(cmd2, stdin=p1.stdout)
|
||||
@@ -209,11 +215,18 @@ def make_ca(ca_dir, ca_name, days=3650, issuing_ca=None, ca_publish_base_url=Non
|
||||
_config.update(patch)
|
||||
return True
|
||||
|
||||
if not issuing_ca:
|
||||
print(f"Root CA already exists in {ca_dir}, skipping.")
|
||||
_config.update({"ca_publish_base_url": ca_publish_base_url} if ca_publish_base_url else {})
|
||||
return True
|
||||
|
||||
issuing_ca_dir = os.path.join(ca_dir, issuing_ca)
|
||||
issuing_ca_cert = os.path.join(issuing_ca_dir, "ca_cert.pem")
|
||||
issuing_ca_key = os.path.join(issuing_ca_dir, "ca_key.pem")
|
||||
|
||||
if not os.path.isfile(issuing_ca_cert) or not os.path.isfile(issuing_ca_key):
|
||||
if os.path.isfile(issuing_ca_cert) and os.path.isfile(issuing_ca_key):
|
||||
print(f"Issuing CA '{issuing_ca}' already exists in {issuing_ca_dir}, skipping.")
|
||||
else:
|
||||
print(f"Generating issuing CA certificate '{ca_name}' and key...")
|
||||
os.makedirs(issuing_ca_dir, exist_ok=True)
|
||||
req_cmd = [
|
||||
@@ -254,11 +267,15 @@ def make_ca(ca_dir, ca_name, days=3650, issuing_ca=None, ca_publish_base_url=Non
|
||||
|
||||
|
||||
def make_cert(cert_subject_name, sans=None, ca_dir=None, cert_dir=None,
|
||||
issuing_ca=None, days=365, ca_publish_base_url=None):
|
||||
issuing_ca=None, days=365, ca_publish_base_url=None, cert_type="server"):
|
||||
if issuing_ca == "ca":
|
||||
_err("--issuing-ca cannot be 'ca' as it is reserved for the root CA.")
|
||||
return False
|
||||
|
||||
if cert_type not in ("server", "user"):
|
||||
_err(f"--type must be 'server' or 'user', got '{cert_type}'.")
|
||||
return False
|
||||
|
||||
if not ca_dir or not os.path.isdir(ca_dir):
|
||||
_err(f"CA directory {ca_dir} does not exist.")
|
||||
return False
|
||||
@@ -267,7 +284,7 @@ def make_cert(cert_subject_name, sans=None, ca_dir=None, cert_dir=None,
|
||||
_err("Subject name is required.")
|
||||
return False
|
||||
|
||||
if not _is_dns(cert_subject_name):
|
||||
if cert_type == "server" and not _is_dns(cert_subject_name):
|
||||
_err(f"Invalid subject name '{cert_subject_name}'. Must be a valid DNS name.")
|
||||
return False
|
||||
|
||||
@@ -297,63 +314,82 @@ def make_cert(cert_subject_name, sans=None, ca_dir=None, cert_dir=None,
|
||||
|
||||
cert_name = cert_subject_name.split(".", 1)[0]
|
||||
|
||||
san_entries = [f"DNS:{cert_subject_name}"]
|
||||
san_entries = [f"DNS:{cert_subject_name}"] if cert_type == "server" else []
|
||||
for entry in sans or []:
|
||||
if _is_ip(entry):
|
||||
try:
|
||||
ipaddress.IPv4Address(entry)
|
||||
except ipaddress.AddressValueError:
|
||||
_err(f"Invalid SAN entry '{entry}'")
|
||||
return False
|
||||
san_entries.append(f"IP:{entry}")
|
||||
elif _is_email(entry):
|
||||
san_entries.append(f"email:{entry}")
|
||||
elif _is_dns(entry):
|
||||
san_entries.append(f"DNS:{entry}")
|
||||
else:
|
||||
_err(f"Invalid SAN entry '{entry}'")
|
||||
return False
|
||||
|
||||
sans_ext = "subjectAltName=" + ",".join(san_entries)
|
||||
|
||||
print(f"Generating server certificate for '{cert_subject_name}' with SANs:")
|
||||
if cert_type == "server":
|
||||
print(f"Generating server certificate for '{cert_subject_name}' with SANs:")
|
||||
else:
|
||||
print(f"Generating user certificate for '{cert_subject_name}':")
|
||||
for san in san_entries:
|
||||
print(f" - {san}")
|
||||
|
||||
cert_out = os.path.join(cert_dir, f"{cert_name}_cert.pem")
|
||||
key_out = os.path.join(cert_dir, f"{cert_name}_key.pem")
|
||||
|
||||
if not os.path.isfile(cert_out) or not os.path.isfile(key_out):
|
||||
print("Generating server certificate and key...")
|
||||
req_cmd = [
|
||||
OPENSSL, "req",
|
||||
"-newkey", "rsa:4096",
|
||||
"-keyout", key_out,
|
||||
"-noenc",
|
||||
"-subj", f"/CN={cert_subject_name}",
|
||||
"-addext", "basicConstraints=critical,CA:FALSE",
|
||||
if os.path.isfile(cert_out) and os.path.isfile(key_out):
|
||||
print(f"Certificate already exists in {cert_dir}, skipping.")
|
||||
return True
|
||||
|
||||
print("Generating certificate and key...")
|
||||
req_cmd = [
|
||||
OPENSSL, "req",
|
||||
"-newkey", "rsa:4096",
|
||||
"-keyout", key_out,
|
||||
"-noenc",
|
||||
"-subj", f"/CN={cert_subject_name}",
|
||||
"-addext", "basicConstraints=critical,CA:FALSE",
|
||||
]
|
||||
if cert_type == "server":
|
||||
req_cmd += [
|
||||
"-addext", "keyUsage=critical,digitalSignature,keyEncipherment",
|
||||
"-addext", "extendedKeyUsage=serverAuth,clientAuth",
|
||||
"-addext", sans_ext,
|
||||
]
|
||||
if aia_url:
|
||||
req_cmd += ["-addext", f"authorityInfoAccess=caIssuers;URI:{aia_url}"]
|
||||
if cdp_url:
|
||||
req_cmd += ["-addext", f"crlDistributionPoints=URI:{cdp_url}"]
|
||||
x509_cmd = [
|
||||
OPENSSL, "x509",
|
||||
"-req",
|
||||
"-CA", ca_cert_path,
|
||||
"-CAkey", ca_key_path,
|
||||
"-copy_extensions", "copyall",
|
||||
"-days", str(days),
|
||||
"-text",
|
||||
"-out", cert_out,
|
||||
else:
|
||||
req_cmd += [
|
||||
"-addext", "keyUsage=critical,digitalSignature,nonRepudiation",
|
||||
"-addext", "extendedKeyUsage=clientAuth,emailProtection,codeSigning",
|
||||
]
|
||||
if not _pipe(req_cmd, x509_cmd):
|
||||
_err("Failed to generate server certificate and key.")
|
||||
return False
|
||||
|
||||
_config.append_history(issuing_ca or "ca", {
|
||||
"name": cert_subject_name,
|
||||
"serial": _read_serial(cert_out),
|
||||
"created": _now(),
|
||||
"expires": _read_expiry(cert_out),
|
||||
})
|
||||
if san_entries:
|
||||
req_cmd += ["-addext", "subjectAltName=" + ",".join(san_entries)]
|
||||
if aia_url:
|
||||
req_cmd += ["-addext", f"authorityInfoAccess=caIssuers;URI:{aia_url}"]
|
||||
if cdp_url:
|
||||
req_cmd += ["-addext", f"crlDistributionPoints=URI:{cdp_url}"]
|
||||
x509_cmd = [
|
||||
OPENSSL, "x509",
|
||||
"-req",
|
||||
"-CA", ca_cert_path,
|
||||
"-CAkey", ca_key_path,
|
||||
"-copy_extensions", "copyall",
|
||||
"-days", str(days),
|
||||
"-text",
|
||||
"-out", cert_out,
|
||||
]
|
||||
if not _pipe(req_cmd, x509_cmd):
|
||||
_err("Failed to generate certificate and key.")
|
||||
return False
|
||||
|
||||
_config.append_history(issuing_ca or "ca", {
|
||||
"name": cert_subject_name,
|
||||
"serial": _read_serial(cert_out),
|
||||
"created": _now(),
|
||||
"expires": _read_expiry(cert_out),
|
||||
})
|
||||
return True
|
||||
|
||||
|
||||
@@ -362,7 +398,7 @@ def make_pfx(cert_path, ca_dir, issuing_ca=None, password=None, apple_openssl=Fa
|
||||
_err("--issuing-ca cannot be 'ca' as it is reserved for the root CA.")
|
||||
return False
|
||||
|
||||
cert_dir = os.path.dirname(cert_path)
|
||||
cert_dir = os.path.dirname(cert_path) or "."
|
||||
cert_basename = os.path.basename(cert_path)
|
||||
cert_name = cert_basename[:-len("_cert.pem")] if cert_basename.endswith("_cert.pem") else cert_basename
|
||||
key_path = os.path.join(cert_dir, f"{cert_name}_key.pem")
|
||||
@@ -380,9 +416,8 @@ def make_pfx(cert_path, ca_dir, issuing_ca=None, password=None, apple_openssl=Fa
|
||||
return False
|
||||
|
||||
root_ca_cert_path = os.path.join(ca_dir, "ca_cert.pem")
|
||||
root_ca_key_path = os.path.join(ca_dir, "ca_key.pem")
|
||||
if not os.path.isfile(root_ca_cert_path) or not os.path.isfile(root_ca_key_path):
|
||||
_err(f"CA certificate or key not found in {ca_dir}.")
|
||||
if not os.path.isfile(root_ca_cert_path):
|
||||
_err(f"Root CA certificate not found in {ca_dir}.")
|
||||
return False
|
||||
|
||||
if issuing_ca:
|
||||
@@ -447,6 +482,10 @@ def make_crl(ca_dir, issuing_ca=None, days=30):
|
||||
e for e in _config.get("history", {}).get(history_key, [])
|
||||
if "revoked" in e and "expires" in e
|
||||
]
|
||||
crl_numbers = dict(_config.get("crl_number", {}))
|
||||
crl_numbers[history_key] = crl_numbers.get(history_key, 0) + 1
|
||||
crl_number_hex = f"{crl_numbers[history_key]:X}"
|
||||
crl_number_hex = crl_number_hex.zfill(len(crl_number_hex) + len(crl_number_hex) % 2)
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
index_txt = os.path.join(tmp, "index.txt")
|
||||
@@ -459,7 +498,7 @@ def make_crl(ca_dir, issuing_ca=None, days=30):
|
||||
revoked_asn1 = _iso_to_asn1(e["revoked"])
|
||||
f.write(f"R\t{expires_asn1}\t{revoked_asn1}\t{e['serial']}\tunknown\t/CN={e['name']}\n")
|
||||
with open(crlnumber, "w") as f:
|
||||
f.write("01\n")
|
||||
f.write(f"{crl_number_hex}\n")
|
||||
with open(cnf_path, "w") as f:
|
||||
f.write(
|
||||
"[ ca ]\ndefault_ca = CA_default\n\n"
|
||||
@@ -478,6 +517,7 @@ def make_crl(ca_dir, issuing_ca=None, days=30):
|
||||
_err("Failed to generate CRL.")
|
||||
return False
|
||||
|
||||
_config.update({"crl_number": crl_numbers})
|
||||
print(f"CRL written to {crl_path}")
|
||||
return True
|
||||
|
||||
@@ -528,6 +568,8 @@ def _build_parser():
|
||||
p_cert.add_argument("--issuing-ca", default=None, help="Specify the issuing CA")
|
||||
p_cert.add_argument("--days", type=int, default=None, help="Validity period in days (default: 365)")
|
||||
p_cert.add_argument("--cert-dir", help="Directory to store the certificate files")
|
||||
p_cert.add_argument("--type", dest="cert_type", choices=["server", "user"], default="server",
|
||||
help="Certificate type (default: server)")
|
||||
p_cert.add_argument("--openssl", default=None, metavar="PATH",
|
||||
help=f"Path to the openssl binary (default: {OPENSSL})")
|
||||
p_cert.add_argument("subject_name", help="Subject name for the certificate")
|
||||
@@ -568,9 +610,14 @@ def main(argv=None):
|
||||
ca_publish_base_url = getattr(args, "ca_publish_base_url", None) or _config.get("ca_publish_base_url")
|
||||
|
||||
days_cfg = _config.get("days", {})
|
||||
days_default = {"make-ca": days_cfg.get("ca", 3650),
|
||||
"make-cert": days_cfg.get("cert", 365),
|
||||
"make-crl": days_cfg.get("crl", 30)}.get(args.command)
|
||||
days = args.days if getattr(args, "days", None) is not None else days_default
|
||||
if days is not None and days < 1:
|
||||
parser.error("--days must be a positive integer")
|
||||
|
||||
if args.command == "make-ca":
|
||||
days = args.days or days_cfg.get("ca", 3650)
|
||||
ok = make_ca(
|
||||
ca_dir, args.ca_name,
|
||||
days=days,
|
||||
@@ -578,7 +625,6 @@ def main(argv=None):
|
||||
ca_publish_base_url=ca_publish_base_url,
|
||||
)
|
||||
elif args.command == "make-cert":
|
||||
days = args.days or days_cfg.get("cert", 365)
|
||||
ok = make_cert(
|
||||
args.subject_name,
|
||||
sans=args.sans,
|
||||
@@ -587,6 +633,7 @@ def main(argv=None):
|
||||
issuing_ca=issuing_ca,
|
||||
days=days,
|
||||
ca_publish_base_url=ca_publish_base_url,
|
||||
cert_type=args.cert_type,
|
||||
)
|
||||
elif args.command == "make-pfx":
|
||||
ok = make_pfx(
|
||||
@@ -596,7 +643,6 @@ def main(argv=None):
|
||||
apple_openssl=args.apple_openssl,
|
||||
)
|
||||
elif args.command == "make-crl":
|
||||
days = args.days or days_cfg.get("crl", 30)
|
||||
ok = make_crl(ca_dir, issuing_ca=issuing_ca, days=days)
|
||||
elif args.command == "revoke-cert":
|
||||
ok = revoke_cert(args.cert_path, ca_dir, issuing_ca=issuing_ca)
|
||||
|
||||
Reference in new issue
Block a user