cryptography

Configuring Certificate Authority with OpenSSL

A Certificate Authority (CA) is the trust anchor in a PKI hierarchy, responsible for issuing, signing, and revoking digital certificates. This skill covers building a two-tier CA hierarchy (Root CA + Intermediate CA) using OpenSSL and the Python cryptography library, including CRL distribution, OCSP responder configuration, and certificate policy management.

certificate-authoritycryptographyopensslpkix509
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

Overview

A Certificate Authority (CA) is the trust anchor in a PKI hierarchy, responsible for issuing, signing, and revoking digital certificates. This skill covers building a two-tier CA hierarchy (Root CA + Intermediate CA) using OpenSSL and the Python cryptography library, including CRL distribution, OCSP responder configuration, and certificate policy management.

When to Use

  • When deploying or configuring configuring certificate authority with openssl capabilities in your environment
  • When establishing security controls aligned to compliance requirements
  • When building or improving security architecture for this domain
  • When conducting security assessments that require this implementation

Prerequisites

  • Familiarity with cryptography concepts and tools
  • Access to a test or lab environment for safe execution
  • Python 3.8+ with required dependencies installed
  • Appropriate authorization for any testing activities

Objectives

  • Create a Root CA with self-signed certificate
  • Create an Intermediate CA signed by the Root CA
  • Issue server and client certificates from the Intermediate CA
  • Configure Certificate Revocation Lists (CRLs)
  • Implement certificate policies and constraints
  • Build a complete PKI hierarchy programmatically

Key Concepts

CA Hierarchy

Root CA (offline, air-gapped)
  |
  +-- Intermediate CA (online, operational)
        |
        +-- Server Certificates
        +-- Client Certificates
        +-- Code Signing Certificates

Certificate Extensions

Extension Purpose Critical
basicConstraints CA:TRUE/FALSE, pathLenConstraint Yes
keyUsage keyCertSign, cRLSign, digitalSignature Yes
extendedKeyUsage serverAuth, clientAuth, codeSigning No
subjectKeyIdentifier Hash of public key No
authorityKeyIdentifier Issuer's key identifier No
crlDistributionPoints URL to CRL No
authorityInfoAccess OCSP responder URL No

Security Considerations

  • Root CA private key must be stored offline (air-gapped HSM)
  • Use minimum 4096-bit RSA or P-384 ECDSA for CA keys
  • Set path length constraints on intermediate CAs
  • Implement certificate policies (OIDs)
  • Enable CRL and OCSP for revocation checking
  • Audit all certificate issuance operations

Validation Criteria

  • Root CA self-signed certificate is valid
  • Intermediate CA certificate chains to Root CA
  • Issued certificates chain to Intermediate -> Root
  • Path length constraints are enforced
  • CRL is generated and accessible
  • Revoked certificates appear in CRL
  • Certificate policies are correctly embedded
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 3

api-reference.md1.7 KB

Certificate Authority with OpenSSL — API Reference

Libraries

Library Install Purpose
cryptography pip install cryptography X.509 certificate generation, parsing, and validation
pyOpenSSL pip install pyOpenSSL OpenSSL wrapper for certificate operations

Key cryptography Methods

Method Description
x509.CertificateBuilder() Build X.509 certificates
rsa.generate_private_key(65537, key_size) Generate RSA private key
x509.load_pem_x509_certificate(data) Parse PEM certificate
cert.subject.rfc4514_string() Get subject as RFC 4514 string
x509.random_serial_number() Generate unique serial number

OpenSSL CLI Commands

Command Purpose
openssl req -x509 -newkey rsa:4096 -sha256 -days 3650 Create self-signed CA
openssl req -new -key server.key -out server.csr Generate CSR
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key Sign certificate
openssl verify -CAfile ca.crt server.crt Verify certificate chain
openssl x509 -in cert.pem -text -noout Display certificate details

Certificate Best Practices

Parameter Recommended Value
Root CA Key Size RSA 4096 or EC P-384
Server Key Size RSA 2048+ or EC P-256
Signature Algorithm SHA-256 or SHA-384
Root CA Validity 10-20 years
Server Cert Validity 1 year (398 days max for public)

External References

standards.md1.2 KB

Standards and References - Certificate Authority with OpenSSL

Primary Standards

RFC 5280 - Internet X.509 PKI Certificate and CRL Profile

RFC 6960 - X.509 OCSP

RFC 3647 - Internet X.509 PKI Certificate Policy and Certification Practices Framework

NIST SP 800-57 Part 1 Rev. 5

CA/Browser Forum Baseline Requirements

Tools

OpenSSL

Python cryptography library

workflows.md2.2 KB

Workflows - Certificate Authority with OpenSSL

Workflow 1: Build Two-Tier CA Hierarchy

[Generate Root CA Key] (RSA 4096 / ECDSA P-384)
      |
[Create Root CA Self-Signed Certificate]
(validity: 20 years, basicConstraints: CA:TRUE)
      |
[Store Root CA Key Offline]
      |
[Generate Intermediate CA Key]
      |
[Create Intermediate CA CSR]
      |
[Sign Intermediate CSR with Root CA]
(pathLenConstraint: 0, keyUsage: keyCertSign, cRLSign)
      |
[Create CA Chain Bundle]
(intermediate.crt + root.crt)

Workflow 2: Issue End-Entity Certificate

[Applicant Generates Key + CSR]
      |
[Submit CSR to Intermediate CA]
      |
[Validate CSR]
(check subject, SAN, key strength)
      |
[Sign with Intermediate CA Key]
(basicConstraints: CA:FALSE)
(extendedKeyUsage: serverAuth / clientAuth)
      |
[Issue Certificate]
      |
[Record in Certificate Database]

Workflow 3: Certificate Revocation

[Revocation Request]
      |
[Verify Authorization]
      |
[Revoke Certificate]
(record serial number + reason + date)
      |
[Generate Updated CRL]
(sign with CA key, set nextUpdate)
      |
[Publish CRL to Distribution Point]
      |
[Update OCSP Responder Database]

Workflow 4: OpenSSL CA Commands

# 1. Create CA directory structure
mkdir -p ca/{certs,crl,newcerts,private}
touch ca/index.txt
echo 1000 > ca/serial
echo 1000 > ca/crlnumber
 
# 2. Generate Root CA
openssl genrsa -aes256 -out ca/private/ca.key 4096
openssl req -config ca/openssl.cnf -key ca/private/ca.key \
    -new -x509 -days 7300 -sha256 -extensions v3_ca -out ca/certs/ca.crt
 
# 3. Generate Intermediate CA
openssl genrsa -aes256 -out intermediate/private/intermediate.key 4096
openssl req -config intermediate/openssl.cnf \
    -key intermediate/private/intermediate.key -new -sha256 -out intermediate/csr/intermediate.csr
openssl ca -config ca/openssl.cnf -extensions v3_intermediate_ca \
    -days 3650 -notext -md sha256 -in intermediate/csr/intermediate.csr \
    -out intermediate/certs/intermediate.crt
 
# 4. Issue server certificate
openssl req -config intermediate/openssl.cnf \
    -key server.key -new -sha256 -out server.csr
openssl ca -config intermediate/openssl.cnf -extensions server_cert \
    -days 365 -notext -md sha256 -in server.csr -out server.crt

Scripts 2

agent.py5.9 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Certificate Authority management agent using OpenSSL and cryptography library."""

import json
import sys
import argparse
from datetime import datetime

try:
    from cryptography import x509
    from cryptography.x509.oid import NameOID
    from cryptography.hazmat.primitives import hashes
    from cryptography.hazmat.primitives.asymmetric import rsa
    from datetime import timedelta
except ImportError:
    print("Install: pip install cryptography")
    sys.exit(1)


def generate_ca_certificate(cn="Internal Root CA", org="Organization", days=3650):
    """Generate a self-signed root CA certificate."""
    key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
    subject = issuer = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, cn),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, org),
    ])
    cert = (x509.CertificateBuilder()
            .subject_name(subject)
            .issuer_name(issuer)
            .public_key(key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(datetime.utcnow())
            .not_valid_after(datetime.utcnow() + timedelta(days=days))
            .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
            .add_extension(x509.KeyUsage(digital_signature=True, key_cert_sign=True,
                                          crl_sign=True, content_commitment=False,
                                          key_encipherment=False, data_encipherment=False,
                                          key_agreement=False, encipher_only=False,
                                          decipher_only=False), critical=True)
            .sign(key, hashes.SHA256()))
    return cert, key


def generate_server_cert(ca_cert, ca_key, cn, san_names, days=365):
    """Generate a server certificate signed by the CA."""
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
    san = x509.SubjectAlternativeName([x509.DNSName(n) for n in san_names])
    cert = (x509.CertificateBuilder()
            .subject_name(subject)
            .issuer_name(ca_cert.subject)
            .public_key(key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(datetime.utcnow())
            .not_valid_after(datetime.utcnow() + timedelta(days=days))
            .add_extension(san, critical=False)
            .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
            .sign(ca_key, hashes.SHA256()))
    return cert, key


def audit_certificate(cert_path):
    """Audit an existing certificate for security issues."""
    with open(cert_path, "rb") as f:
        cert_data = f.read()
    cert = x509.load_pem_x509_certificate(cert_data)
    issues = []
    if cert.public_key().key_size < 2048:
        issues.append({"issue": "Weak key size", "severity": "HIGH",
                        "detail": f"Key size {cert.public_key().key_size} < 2048"})
    if cert.not_valid_after_utc.replace(tzinfo=None) < datetime.utcnow():
        issues.append({"issue": "Certificate expired", "severity": "CRITICAL",
                        "detail": f"Expired on {cert.not_valid_after_utc}"})
    days_remaining = (cert.not_valid_after_utc.replace(tzinfo=None) - datetime.utcnow()).days
    if 0 < days_remaining < 30:
        issues.append({"issue": "Certificate expiring soon", "severity": "HIGH",
                        "detail": f"{days_remaining} days remaining"})
    sig_algo = cert.signature_hash_algorithm
    if sig_algo and sig_algo.name in ("sha1", "md5"):
        issues.append({"issue": f"Weak signature algorithm: {sig_algo.name}", "severity": "HIGH"})
    return {
        "subject": cert.subject.rfc4514_string(),
        "issuer": cert.issuer.rfc4514_string(),
        "not_before": str(cert.not_valid_before_utc),
        "not_after": str(cert.not_valid_after_utc),
        "key_size": cert.public_key().key_size,
        "serial": cert.serial_number,
        "issues": issues,
    }


def run_audit(cert_path=None, generate=False, cn=None, org=None):
    """Execute CA operations."""
    print(f"\n{'='*60}")
    print(f"  CERTIFICATE AUTHORITY AGENT")
    print(f"  Generated: {datetime.utcnow().isoformat()} UTC")
    print(f"{'='*60}\n")

    if cert_path:
        info = audit_certificate(cert_path)
        print(f"--- CERTIFICATE AUDIT ---")
        print(f"  Subject: {info['subject']}")
        print(f"  Issuer: {info['issuer']}")
        print(f"  Valid: {info['not_before']} to {info['not_after']}")
        print(f"  Key Size: {info['key_size']}")
        print(f"  Issues: {len(info['issues'])}")
        for i in info["issues"]:
            print(f"    [{i['severity']}] {i['issue']}: {i.get('detail', '')}")
        return info

    if generate:
        cert, key = generate_ca_certificate(cn=cn or "Internal Root CA", org=org or "Org")
        print(f"--- CA CERTIFICATE GENERATED ---")
        print(f"  Subject: {cert.subject.rfc4514_string()}")
        print(f"  Serial: {cert.serial_number}")
        return {"status": "generated", "subject": cert.subject.rfc4514_string()}

    return {}


def main():
    parser = argparse.ArgumentParser(description="Certificate Authority Agent")
    parser.add_argument("--audit", help="Path to PEM certificate to audit")
    parser.add_argument("--generate", action="store_true", help="Generate new CA")
    parser.add_argument("--cn", help="Common name for CA cert")
    parser.add_argument("--org", help="Organization name")
    parser.add_argument("--output", help="Save report to JSON file")
    args = parser.parse_args()

    report = run_audit(args.audit, args.generate, args.cn, args.org)
    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2, default=str)
        print(f"\n[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()
process.py16.9 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""
Certificate Authority Builder using Python cryptography library.

Builds a complete two-tier CA hierarchy (Root CA + Intermediate CA)
and issues server/client certificates programmatically.

Requirements:
    pip install cryptography

Usage:
    python process.py build-ca --output ./pki --org "My Organization"
    python process.py issue-cert --ca-dir ./pki --domain server.example.com --type server
    python process.py revoke --ca-dir ./pki --serial 1001
    python process.py generate-crl --ca-dir ./pki
"""

import os
import sys
import json
import argparse
import logging
import datetime
from pathlib import Path
from typing import Dict, Optional, List

from cryptography import x509
from cryptography.x509.oid import NameOID, ExtensionOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, ec
from cryptography.hazmat.backends import default_backend

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def generate_key(key_type: str = "rsa", key_size: int = 4096):
    """Generate a private key."""
    if key_type == "ecdsa":
        return ec.generate_private_key(ec.SECP384R1(), default_backend())
    return rsa.generate_private_key(
        public_exponent=65537, key_size=key_size, backend=default_backend()
    )


def save_key(key, path: Path, passphrase: Optional[str] = None):
    """Save a private key to PEM file."""
    if passphrase:
        enc = serialization.BestAvailableEncryption(passphrase.encode())
    else:
        enc = serialization.NoEncryption()
    path.write_bytes(key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=enc,
    ))


def save_cert(cert, path: Path):
    """Save a certificate to PEM file."""
    path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))


def build_root_ca(
    output_dir: Path,
    organization: str,
    country: str = "US",
    validity_years: int = 20,
) -> Dict:
    """Build a Root CA with self-signed certificate."""
    ca_dir = output_dir / "root-ca"
    ca_dir.mkdir(parents=True, exist_ok=True)
    (ca_dir / "certs").mkdir(exist_ok=True)
    (ca_dir / "private").mkdir(exist_ok=True)

    key = generate_key("rsa", 4096)
    save_key(key, ca_dir / "private" / "root-ca.key")

    subject = x509.Name([
        x509.NameAttribute(NameOID.COUNTRY_NAME, country),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization),
        x509.NameAttribute(NameOID.COMMON_NAME, f"{organization} Root CA"),
    ])

    now = datetime.datetime.utcnow()
    cert = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(subject)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now)
        .not_valid_after(now + datetime.timedelta(days=validity_years * 365))
        .add_extension(
            x509.BasicConstraints(ca=True, path_length=1), critical=True
        )
        .add_extension(
            x509.KeyUsage(
                digital_signature=True, key_cert_sign=True, crl_sign=True,
                content_commitment=False, key_encipherment=False,
                data_encipherment=False, key_agreement=False,
                encipher_only=False, decipher_only=False,
            ),
            critical=True,
        )
        .add_extension(
            x509.SubjectKeyIdentifier.from_public_key(key.public_key()),
            critical=False,
        )
        .sign(key, hashes.SHA384(), default_backend())
    )

    save_cert(cert, ca_dir / "certs" / "root-ca.crt")

    # Initialize serial number tracker
    (ca_dir / "serial.json").write_text(json.dumps({"next_serial": 1001}))
    (ca_dir / "index.json").write_text(json.dumps({"certificates": []}))

    logger.info(f"Root CA created: {ca_dir}")
    return {
        "type": "root-ca",
        "subject": subject.rfc4514_string(),
        "key_path": str(ca_dir / "private" / "root-ca.key"),
        "cert_path": str(ca_dir / "certs" / "root-ca.crt"),
        "serial_number": hex(cert.serial_number),
        "valid_until": cert.not_valid_after_utc.isoformat(),
    }


def build_intermediate_ca(
    output_dir: Path,
    organization: str,
    country: str = "US",
    validity_years: int = 10,
) -> Dict:
    """Build an Intermediate CA signed by the Root CA."""
    root_dir = output_dir / "root-ca"
    int_dir = output_dir / "intermediate-ca"
    int_dir.mkdir(parents=True, exist_ok=True)
    (int_dir / "certs").mkdir(exist_ok=True)
    (int_dir / "private").mkdir(exist_ok=True)

    root_key_data = (root_dir / "private" / "root-ca.key").read_bytes()
    root_key = serialization.load_pem_private_key(root_key_data, password=None, backend=default_backend())

    root_cert_data = (root_dir / "certs" / "root-ca.crt").read_bytes()
    root_cert = x509.load_pem_x509_certificate(root_cert_data, default_backend())

    int_key = generate_key("rsa", 4096)
    save_key(int_key, int_dir / "private" / "intermediate-ca.key")

    subject = x509.Name([
        x509.NameAttribute(NameOID.COUNTRY_NAME, country),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, organization),
        x509.NameAttribute(NameOID.COMMON_NAME, f"{organization} Intermediate CA"),
    ])

    now = datetime.datetime.utcnow()
    cert = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(root_cert.subject)
        .public_key(int_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now)
        .not_valid_after(now + datetime.timedelta(days=validity_years * 365))
        .add_extension(
            x509.BasicConstraints(ca=True, path_length=0), critical=True
        )
        .add_extension(
            x509.KeyUsage(
                digital_signature=True, key_cert_sign=True, crl_sign=True,
                content_commitment=False, key_encipherment=False,
                data_encipherment=False, key_agreement=False,
                encipher_only=False, decipher_only=False,
            ),
            critical=True,
        )
        .add_extension(
            x509.SubjectKeyIdentifier.from_public_key(int_key.public_key()),
            critical=False,
        )
        .add_extension(
            x509.AuthorityKeyIdentifier.from_issuer_public_key(root_key.public_key()),
            critical=False,
        )
        .sign(root_key, hashes.SHA384(), default_backend())
    )

    save_cert(cert, int_dir / "certs" / "intermediate-ca.crt")

    # Create chain file
    chain = cert.public_bytes(serialization.Encoding.PEM) + root_cert.public_bytes(serialization.Encoding.PEM)
    (int_dir / "certs" / "ca-chain.crt").write_bytes(chain)

    # Initialize serial and index
    (int_dir / "serial.json").write_text(json.dumps({"next_serial": 2001}))
    (int_dir / "index.json").write_text(json.dumps({"certificates": [], "revoked": []}))

    logger.info(f"Intermediate CA created: {int_dir}")
    return {
        "type": "intermediate-ca",
        "subject": subject.rfc4514_string(),
        "key_path": str(int_dir / "private" / "intermediate-ca.key"),
        "cert_path": str(int_dir / "certs" / "intermediate-ca.crt"),
        "chain_path": str(int_dir / "certs" / "ca-chain.crt"),
        "serial_number": hex(cert.serial_number),
        "valid_until": cert.not_valid_after_utc.isoformat(),
    }


def issue_certificate(
    ca_dir: Path,
    domain: str,
    cert_type: str = "server",
    validity_days: int = 365,
    san_domains: Optional[List[str]] = None,
) -> Dict:
    """Issue a certificate from the Intermediate CA."""
    int_dir = ca_dir / "intermediate-ca"

    int_key_data = (int_dir / "private" / "intermediate-ca.key").read_bytes()
    int_key = serialization.load_pem_private_key(int_key_data, password=None, backend=default_backend())

    int_cert_data = (int_dir / "certs" / "intermediate-ca.crt").read_bytes()
    int_cert = x509.load_pem_x509_certificate(int_cert_data, default_backend())

    # Read and update serial
    serial_data = json.loads((int_dir / "serial.json").read_text())
    serial_num = serial_data["next_serial"]
    serial_data["next_serial"] = serial_num + 1
    (int_dir / "serial.json").write_text(json.dumps(serial_data))

    # Generate end-entity key
    ee_key = generate_key("ecdsa")
    subject = x509.Name([
        x509.NameAttribute(NameOID.COMMON_NAME, domain),
    ])

    san_list = [x509.DNSName(domain)]
    if san_domains:
        for d in san_domains:
            san_list.append(x509.DNSName(d))

    if cert_type == "server":
        eku = x509.ExtendedKeyUsage([x509.oid.ExtendedKeyUsageOID.SERVER_AUTH])
    elif cert_type == "client":
        eku = x509.ExtendedKeyUsage([x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH])
    else:
        eku = x509.ExtendedKeyUsage([
            x509.oid.ExtendedKeyUsageOID.SERVER_AUTH,
            x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH,
        ])

    now = datetime.datetime.utcnow()
    cert = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(int_cert.subject)
        .public_key(ee_key.public_key())
        .serial_number(serial_num)
        .not_valid_before(now)
        .not_valid_after(now + datetime.timedelta(days=validity_days))
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
        .add_extension(
            x509.KeyUsage(
                digital_signature=True, key_encipherment=True,
                key_cert_sign=False, crl_sign=False,
                content_commitment=False, data_encipherment=False,
                key_agreement=False, encipher_only=False, decipher_only=False,
            ),
            critical=True,
        )
        .add_extension(eku, critical=False)
        .add_extension(x509.SubjectAlternativeName(san_list), critical=False)
        .add_extension(
            x509.AuthorityKeyIdentifier.from_issuer_public_key(int_key.public_key()),
            critical=False,
        )
        .sign(int_key, hashes.SHA256(), default_backend())
    )

    certs_dir = int_dir / "certs" / "issued"
    certs_dir.mkdir(exist_ok=True)

    save_key(ee_key, certs_dir / f"{domain}.key")
    save_cert(cert, certs_dir / f"{domain}.crt")

    # Update index
    index_data = json.loads((int_dir / "index.json").read_text())
    index_data["certificates"].append({
        "serial": serial_num,
        "domain": domain,
        "type": cert_type,
        "issued": now.isoformat(),
        "expires": (now + datetime.timedelta(days=validity_days)).isoformat(),
        "status": "valid",
    })
    (int_dir / "index.json").write_text(json.dumps(index_data, indent=2))

    logger.info(f"Issued {cert_type} certificate for {domain} (serial: {serial_num})")
    return {
        "domain": domain,
        "serial": serial_num,
        "type": cert_type,
        "key_path": str(certs_dir / f"{domain}.key"),
        "cert_path": str(certs_dir / f"{domain}.crt"),
        "chain_path": str(int_dir / "certs" / "ca-chain.crt"),
        "valid_until": cert.not_valid_after_utc.isoformat(),
    }


def revoke_certificate(ca_dir: Path, serial: int, reason: str = "unspecified") -> Dict:
    """Revoke a certificate by serial number."""
    int_dir = ca_dir / "intermediate-ca"
    index_data = json.loads((int_dir / "index.json").read_text())

    reason_map = {
        "unspecified": x509.ReasonFlags.unspecified,
        "key_compromise": x509.ReasonFlags.key_compromise,
        "ca_compromise": x509.ReasonFlags.ca_compromise,
        "affiliation_changed": x509.ReasonFlags.affiliation_changed,
        "superseded": x509.ReasonFlags.superseded,
        "cessation_of_operation": x509.ReasonFlags.cessation_of_operation,
    }

    found = False
    for cert_entry in index_data["certificates"]:
        if cert_entry["serial"] == serial:
            cert_entry["status"] = "revoked"
            cert_entry["revoked_at"] = datetime.datetime.utcnow().isoformat()
            cert_entry["revocation_reason"] = reason
            found = True
            break

    if not found:
        raise ValueError(f"Certificate with serial {serial} not found")

    if "revoked" not in index_data:
        index_data["revoked"] = []

    index_data["revoked"].append({
        "serial": serial,
        "revoked_at": datetime.datetime.utcnow().isoformat(),
        "reason": reason,
    })

    (int_dir / "index.json").write_text(json.dumps(index_data, indent=2))
    logger.info(f"Revoked certificate serial {serial} (reason: {reason})")

    return {"serial": serial, "status": "revoked", "reason": reason}


def generate_crl(ca_dir: Path, validity_days: int = 30) -> Dict:
    """Generate a Certificate Revocation List."""
    int_dir = ca_dir / "intermediate-ca"

    int_key_data = (int_dir / "private" / "intermediate-ca.key").read_bytes()
    int_key = serialization.load_pem_private_key(int_key_data, password=None, backend=default_backend())

    int_cert_data = (int_dir / "certs" / "intermediate-ca.crt").read_bytes()
    int_cert = x509.load_pem_x509_certificate(int_cert_data, default_backend())

    index_data = json.loads((int_dir / "index.json").read_text())

    now = datetime.datetime.utcnow()
    builder = x509.CertificateRevocationListBuilder()
    builder = builder.issuer_name(int_cert.subject)
    builder = builder.last_update(now)
    builder = builder.next_update(now + datetime.timedelta(days=validity_days))

    reason_map = {
        "unspecified": x509.ReasonFlags.unspecified,
        "key_compromise": x509.ReasonFlags.key_compromise,
        "ca_compromise": x509.ReasonFlags.ca_compromise,
        "superseded": x509.ReasonFlags.superseded,
        "cessation_of_operation": x509.ReasonFlags.cessation_of_operation,
    }

    for entry in index_data.get("revoked", []):
        revoked_cert = (
            x509.RevokedCertificateBuilder()
            .serial_number(entry["serial"])
            .revocation_date(datetime.datetime.fromisoformat(entry["revoked_at"]))
            .add_extension(
                x509.CRLReason(reason_map.get(entry.get("reason", "unspecified"), x509.ReasonFlags.unspecified)),
                critical=False,
            )
            .build()
        )
        builder = builder.add_revoked_certificate(revoked_cert)

    crl = builder.sign(int_key, hashes.SHA256(), default_backend())

    crl_path = int_dir / "crl" / "intermediate.crl"
    crl_path.parent.mkdir(exist_ok=True)
    crl_path.write_bytes(crl.public_bytes(serialization.Encoding.PEM))

    logger.info(f"Generated CRL with {len(index_data.get('revoked', []))} revoked certificates")
    return {
        "crl_path": str(crl_path),
        "revoked_count": len(index_data.get("revoked", [])),
        "next_update": (now + datetime.timedelta(days=validity_days)).isoformat(),
    }


def main():
    parser = argparse.ArgumentParser(description="Certificate Authority Builder")
    subparsers = parser.add_subparsers(dest="command")

    build = subparsers.add_parser("build-ca", help="Build complete CA hierarchy")
    build.add_argument("--output", "-o", default="./pki", help="Output directory")
    build.add_argument("--org", required=True, help="Organization name")
    build.add_argument("--country", default="US", help="Country code")

    issue = subparsers.add_parser("issue-cert", help="Issue a certificate")
    issue.add_argument("--ca-dir", required=True, help="CA directory")
    issue.add_argument("--domain", required=True, help="Domain name")
    issue.add_argument("--type", choices=["server", "client", "both"], default="server")
    issue.add_argument("--days", type=int, default=365, help="Validity days")
    issue.add_argument("--san", nargs="*", help="Additional SAN domains")

    rev = subparsers.add_parser("revoke", help="Revoke a certificate")
    rev.add_argument("--ca-dir", required=True, help="CA directory")
    rev.add_argument("--serial", type=int, required=True, help="Serial number")
    rev.add_argument("--reason", default="unspecified", help="Revocation reason")

    crl = subparsers.add_parser("generate-crl", help="Generate CRL")
    crl.add_argument("--ca-dir", required=True, help="CA directory")
    crl.add_argument("--days", type=int, default=30, help="CRL validity days")

    args = parser.parse_args()

    if args.command == "build-ca":
        output = Path(args.output)
        root = build_root_ca(output, args.org, args.country)
        intermediate = build_intermediate_ca(output, args.org, args.country)
        print(json.dumps({"root_ca": root, "intermediate_ca": intermediate}, indent=2))
    elif args.command == "issue-cert":
        result = issue_certificate(
            Path(args.ca_dir), args.domain, args.type, args.days, args.san
        )
        print(json.dumps(result, indent=2))
    elif args.command == "revoke":
        result = revoke_certificate(Path(args.ca_dir), args.serial, args.reason)
        print(json.dumps(result, indent=2))
    elif args.command == "generate-crl":
        result = generate_crl(Path(args.ca_dir), args.days)
        print(json.dumps(result, indent=2))
    else:
        parser.print_help()


if __name__ == "__main__":
    main()

Assets 1

template.mdtext/markdown · 1.7 KB
Keep exploring